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 extent store of the
28//! design's Layer 1: a slot in a pool-owned anonymous-memory extent arena
29//! holding the chunk's lz4-compressed bytes, later paged to swap or demoted
30//! to file per the pool's [`ExtentBackend`].
31//!
32//! Memory descends a ladder of tiers, each with its own ceiling and each
33//! cheaper to vacate than the one above:
34//!
35//! * **Slots** (uncompressed, free reads) — bounded by the budget; crossing
36//!   it compresses the oldest chunks into extents and releases their slots.
37//! * **Warm free slots** (pages kept for fault-free reuse) — bounded by the
38//!   warm cap.
39//! * **Compressed-resident extents** (reads decompress, no device) — bounded
40//!   by the headroom the RSS target leaves above the first two; crossing it
41//!   pushes the oldest extents to the swap device with `MADV_PAGEOUT`.
42//! * **The swap device** (overflow). Reads fault and decompress. A pool
43//!   built over [`ExtentBackend::File`] demotes extents to files in a
44//!   scratch directory instead, and reads them back with `pread`.
45//!
46//! Residency is a state, not a type. It descends through eviction and
47//! ascends through exactly one transition: an admitting read
48//! ([`ChunkHandle::read_into_admit`]) lifts an evicted chunk back to
49//! `BackedResident` when a slot is free within the budget or stealable from
50//! a clean backed victim of the same class, never by evicting or
51//! compressing anything. Plain reads ([`ChunkHandle::read_into`]) leave
52//! residency untouched. Eviction I/O runs on spill threads when enabled —
53//! `WriteInFlight` marks a chunk whose compression a spill thread owns — and
54//! inline on the evicting caller otherwise. Chunks are immutable after
55//! [`Pool::insert_with`], which is what makes a `BackedResident` slot always
56//! identical to its extent and its eviction free of I/O.
57//!
58//! Freeing an `UnbackedResident` chunk is a pure memory operation — the
59//! design's "never write dead data" win, surfaced as `writes_elided` in
60//! [`PoolStats`]. Budget pressure evicts cold chunks via second-chance
61//! FIFOs banded by the caller-supplied generational depth ([`ChunkHints`]).
62
63mod extent;
64mod file;
65mod region;
66#[cfg(test)]
67mod testing;
68
69use std::collections::VecDeque;
70use std::ops::Range;
71use std::path::PathBuf;
72use std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
73use std::sync::{Arc, Mutex, MutexGuard, Weak};
74
75use crate::cast::CastFrom;
76use crate::pool::extent::{Extent, ExtentArena, Scratch};
77pub use crate::pool::file::READ_LATENCY_BUCKETS;
78use crate::pool::file::{FileStore, IoMode, WriteError};
79use crate::pool::region::{Region, SIZE_CLASSES};
80
81/// Virtual reservation per size class. Purely virtual: physical memory
82/// materializes only for slots in use, and slots are scoped to residency,
83/// so this must exceed the largest plausible *resident* set per class, the
84/// budget plus in-flight slack, not the backlog. It is deliberately enormous
85/// (address space costs nothing, and touched pages are bounded by peak
86/// residency) so that no realistic budget, on any machine size, reaches the
87/// heap-fallback path.
88///
89/// NOTE: Seen OoMs with Miri since it actually allocates the capacity.
90const CLASS_CAPACITY_BYTES: usize = if cfg!(miri) { 16 << 20 } else { 1 << 40 };
91
92/// Bytes per word of a chunk body. Chunk lengths are in `u64` words.
93const WORD_BYTES: usize = std::mem::size_of::<u64>();
94
95/// A chunk-provided transform between a chunk's body bytes and the stored
96/// bytes its extent holds. The pool owns scheduling: spill threads, the
97/// residency state machine, cancellation, and the ledger. It invokes the
98/// codec on opaque bytes at the extent boundary, `encode` when backing a
99/// chunk (on a spill thread, or inline under overload) and `decode` when
100/// reading an evicted one, possibly under the chunk's state lock. The pool
101/// itself has no opinion on the stored form: framing, compression, and
102/// validation all belong to the codec.
103///
104/// Implementations must be pure transforms: no locking, no calls back into
105/// the pool (the state lock may be held at `decode` sites), and no panic on
106/// bytes their own `encode` produced. `decode` must exactly invert
107/// `encode`, and `encode`'s output must never exceed
108/// [`max_stored_len`]`(body.len())`, the bound the extent store's size
109/// classes are provisioned to.
110pub trait ExtentCodec: std::fmt::Debug + Send + Sync {
111    /// Transforms `body` into its stored form, replacing `out`'s contents.
112    /// `out`'s capacity is reused across calls; implementations size it
113    /// themselves.
114    fn encode(&self, body: &[u8], out: &mut Vec<u8>);
115
116    /// Inverts [`ExtentCodec::encode`]: reconstructs into `body` exactly
117    /// the bytes whose encoding produced `stored`. `body` is exactly the
118    /// original body's length, and implementations must panic on a length
119    /// mismatch rather than truncate or pad.
120    fn decode(&self, stored: &[u8], body: &mut [u8]);
121
122    /// Whether the stored form is the body itself, for splitting the pool's
123    /// write-bytes counters by codec.
124    fn is_identity(&self) -> bool {
125        false
126    }
127}
128
129/// The identity [`ExtentCodec`]: the stored form is the body. Encode and
130/// decode are copies, and range reads copy the range directly, so a chunk
131/// stored under this codec pays no compression work in either direction
132/// while remaining fully budgeted and backed like any other extent.
133#[derive(Debug)]
134pub struct IdentityCodec;
135
136/// The [`IdentityCodec`] instance to pass to [`Pool::insert_with`].
137pub static IDENTITY_CODEC: IdentityCodec = IdentityCodec;
138
139impl ExtentCodec for IdentityCodec {
140    fn encode(&self, body: &[u8], out: &mut Vec<u8>) {
141        out.clear();
142        out.extend_from_slice(body);
143    }
144
145    fn decode(&self, stored: &[u8], body: &mut [u8]) {
146        assert_eq!(stored.len(), body.len(), "identity stored form is the body");
147        body.copy_from_slice(stored);
148    }
149
150    fn is_identity(&self) -> bool {
151        true
152    }
153}
154
155/// The extent store a pool is built over, fixed for the pool's lifetime.
156#[derive(Debug, Clone, PartialEq, Eq)]
157pub enum ExtentBackend {
158    /// Compressed extents in the arena, paged to the swap device.
159    Swap,
160    /// Compressed extents in the arena, demoted to files under `dir`.
161    File {
162        /// A directory on a disk-backed filesystem. The files have no name
163        /// and vanish with the process.
164        dir: PathBuf,
165        /// Capacity of the files in bytes. `None` derives it from the
166        /// volume's free space.
167        capacity_bytes: Option<u64>,
168    },
169}
170
171/// The store a pool actually runs on.
172#[derive(Debug, Clone, Copy, PartialEq, Eq)]
173pub enum BackendKind {
174    /// The swap device.
175    Swap,
176    /// Files with `O_DIRECT` I/O.
177    FileDirect,
178    /// Files with page-cache I/O, written back and dropped after each write.
179    FileBuffered,
180}
181
182/// The extent store behind a [`PoolInner`].
183#[derive(Debug)]
184enum ExtentStore {
185    Swap,
186    File(Arc<FileStore>),
187}
188
189/// A chunk body the pool holds in its owner's in-memory form, serializing it
190/// only when it has to leave that form: on eviction, on backing, and on a
191/// copy-out read. Under budget a typed chunk therefore costs the pool its
192/// bookkeeping and nothing else, and [`ChunkHandle::take_typed`] hands the
193/// body back without a copy.
194///
195/// The body is immutable once inserted. Spill threads serialize and drop it
196/// under the chunk's state lock, so it must be `Send`.
197pub trait TypedBody: std::any::Any + Send + std::fmt::Debug {
198    /// The serialized length in `u64` words. Must not change while the pool
199    /// holds the body.
200    fn words(&self) -> usize;
201
202    /// Writes the serialized form into `dst`, which is exactly
203    /// [`TypedBody::words`] long, overwriting all of it.
204    fn serialize_into(&self, dst: &mut [u64]);
205
206    /// The body as [`std::any::Any`], for the owner to downcast a borrow.
207    fn as_any(&self) -> &dyn std::any::Any;
208
209    /// The body as [`std::any::Any`], for the owner to downcast a take.
210    fn into_any(self: Box<Self>) -> Box<dyn std::any::Any>;
211}
212
213/// The serialized form of a typed body, in a fresh buffer.
214fn serialize_typed(body: &dyn TypedBody) -> Vec<u64> {
215    let mut words = vec![0; body.words()];
216    body.serialize_into(&mut words);
217    words
218}
219
220/// The largest stored form [`ExtentCodec::encode`] may produce for a
221/// `body_len`-byte body: an incompressible-input expansion matching lz4's
222/// worst case plus a four-byte length prefix. The extent store's size-class
223/// ladder is provisioned to this bound, so a codec that exceeds it can
224/// strand payloads with no class to hold them (they degrade to unpageable
225/// heap fallbacks).
226pub fn max_stored_len(body_len: usize) -> usize {
227    4 + body_len + body_len / 255 + 16
228}
229
230/// Advisory placement hints for a chunk, supplied at insert and immutable
231/// thereafter (merges mint new chunks, so a chunk's generation never
232/// changes). Hints steer policy — eviction order and write-behind
233/// candidacy — never correctness: a mislabeled chunk performs worse, while
234/// the budget and residency invariants hold regardless.
235#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
236pub struct ChunkHints {
237    /// Generational depth of the chunk in its producer's merge structure,
238    /// 0 for the youngest generation (and the unannotated default). Deeper
239    /// chunks are treated as colder: preferred write-behind candidates and
240    /// preferred eviction victims, cheap to evict once backed.
241    pub depth: u8,
242}
243
244/// Number of depth bands the eviction queues are split into; depths at or
245/// beyond the last band share it.
246const DEPTH_BANDS: usize = 4;
247
248/// The eviction-queue band for a chunk of `depth`.
249fn band(depth: u8) -> usize {
250    usize::from(depth).min(DEPTH_BANDS - 1)
251}
252
253/// Resident bytes insertions may reserve above `budget` while enforcement
254/// catches up. Slots can transiently hold this much past the budget, so the
255/// RSS target reserves it too.
256fn insert_slack(budget: u64) -> u64 {
257    budget / 8
258}
259
260/// Residency state of a chunk.
261#[derive(Debug, Clone, Copy, PartialEq, Eq)]
262enum Residency {
263    /// Lives only in the pool; no extent copy exists. Freeing it never
264    /// touches the backing store.
265    UnbackedResident,
266    /// Resident, and an identical extent copy exists; eviction releases
267    /// physical pages without I/O.
268    BackedResident,
269    /// Resident and readable, with compression into an extent scheduled on a
270    /// spill thread. Completion moves an evicting chunk to
271    /// [`Residency::Evicted`] and an eagerly backed one to
272    /// [`Residency::BackedResident`]; a free observed at dequeue cancels the
273    /// write instead.
274    WriteInFlight,
275    /// Extent copy only. The chunk holds no slot. The extent itself may be
276    /// in the arena (the compressed tier), on the swap device, or on file.
277    /// Reads decompress the extent straight into the caller's
278    /// buffer and leave the chunk evicted, except that an admitting read
279    /// may lift it back to [`Residency::BackedResident`].
280    Evicted,
281    /// Larger than the largest size class; held as a plain heap allocation,
282    /// always resident. A prototype limitation, not a design state.
283    Oversize,
284}
285
286/// Snapshot of pool counters.
287#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
288pub struct PoolStats {
289    /// Chunks inserted.
290    pub inserts: u64,
291    /// Inserts written directly to an extent because resident admission was full.
292    pub direct_extent_inserts: u64,
293    /// Chunks freed (handle dropped).
294    pub frees: u64,
295    /// Backing writes elided: chunks dead before their compression
296    /// completed, so no extent write happened. Covers chunks freed while
297    /// `UnbackedResident` and chunks freed while queued for a spill thread
298    /// that had not yet compressed them.
299    pub writes_elided: u64,
300    /// Evictions that compressed the chunk into a new extent.
301    pub evictions_compress: u64,
302    /// Evictions of `BackedResident` chunks: pure page release, no I/O.
303    pub evictions_cheap: u64,
304    /// Compressed bytes written into extents.
305    pub extent_bytes_written: u64,
306    /// Evictions handed to spill threads.
307    pub spill_scheduled: u64,
308    /// Compressions cancelled because the chunk was freed while queued or
309    /// in flight, whatever scheduled them. Eager-backing work counts here
310    /// but never in `spill_scheduled`, so this can exceed that counter.
311    pub spill_cancelled: u64,
312    /// Entries currently queued for or being processed by spill threads.
313    pub spill_in_flight: u64,
314    /// Inserts that fell back to the heap because their size class had no
315    /// free slot (the live set outgrew the class reservation). Heap-backed
316    /// chunks behave like oversize ones: always resident, never paged.
317    pub slot_exhausted_fallbacks: u64,
318    /// Inserts whose payload exceeded the largest size class and therefore
319    /// went straight to a heap-backed oversize chunk.
320    pub oversize_payloads: u64,
321    /// Live size-classed chunks across all classes, whatever their residency.
322    /// For backlog-shaped consumers this tracks the un-drained backlog in
323    /// chunks.
324    pub live_chunks: u64,
325    /// Uncompressed bytes of currently resident chunks (including oversize).
326    pub resident_bytes: u64,
327    /// Uncompressed bytes of live oversize chunks.
328    pub oversize_bytes: u64,
329    /// Class bytes of free slots currently kept warm (pages resident for
330    /// fault-free reuse). Bounded by a fraction of the budget; RSS exceeds
331    /// `resident_bytes` by up to this amount.
332    pub warm_bytes: u64,
333    /// Slot allocations served from the warm list: reuses that faulted no
334    /// pages and skipped the kernel's page zeroing.
335    pub warm_reuses: u64,
336    /// Chunks eagerly compressed to `BackedResident` by idle spill threads
337    /// (write-behind): still readable in their slots, with eviction
338    /// pre-paid.
339    pub eager_backs: u64,
340    /// Evicted chunks re-admitted to `BackedResident` by an admitting read
341    /// out of free budget headroom.
342    pub admissions_budget: u64,
343    /// Evicted chunks re-admitted to `BackedResident` by an admitting read
344    /// stealing the slot of a clean backed victim of the same size class.
345    /// The victim becomes `Evicted` with zero I/O and its extent intact.
346    pub admissions_steal: u64,
347    /// Admitting reads of evicted chunks served as a plain decompress
348    /// instead: no budget headroom (or an exhausted size class), and no
349    /// clean victim whose growth the budget could absorb.
350    pub admissions_denied: u64,
351    /// Allocation bytes of compressed extents currently resident — the
352    /// compressed-but-resident middle tier. Bounded by the RSS target;
353    /// exceeding it pages the oldest extents out to the swap device.
354    pub extent_resident_bytes: u64,
355    /// Allocation bytes of resident extents the RSS target cannot push out:
356    /// retry-capped arena extents (the kernel declined the reclaim advice
357    /// until the retry budget ran out) and heap-fallback extents. The
358    /// compressed tier settles above its capacity by this amount.
359    pub extent_unreclaimable_bytes: u64,
360    /// Extents pushed out of RAM by RSS-target enforcement: to the swap
361    /// device with the whole range observed nonresident afterwards, or
362    /// demoted to file.
363    pub extent_pageouts: u64,
364    /// Pageout passes whose observation found some of the extent's pages
365    /// still mapped: `MADV_PAGEOUT` may decline pages and still succeed, so
366    /// the page table decides. The extent keeps its full resident
367    /// accounting and is retried until its per-extent retry cap. Climbing
368    /// steadily on a loaded pool means pages cannot actually be unmapped to
369    /// the swap device (no swap, or a cgroup that cannot reclaim).
370    pub extent_pageout_incomplete: u64,
371    /// Extent writes that fell back to the heap because their extent-arena
372    /// class had no free slot. Heap-backed extents stay readable but are
373    /// never paged out, so their compressed bytes hold RAM until freed.
374    pub extent_arena_fallbacks: u64,
375    /// Extents freed while resident in a reclaimable arena slot, so the
376    /// free needed no swap or file I/O. In swap mode this includes extents
377    /// paged out and then read back before the free, since the read made
378    /// them resident again.
379    pub extent_demotions_elided: u64,
380    /// Slot bytes of live extents demoted to file. The file fields stay zero
381    /// on a swap-backed pool.
382    pub extent_file_bytes: u64,
383    /// Bytes the file store holds on the filesystem: slots in use plus free
384    /// slots whose blocks were not punched.
385    pub extent_file_allocated_bytes: u64,
386    /// The file store's effective capacity in bytes.
387    pub extent_file_capacity_bytes: u64,
388    /// Demotion writes that reached the device without error, including
389    /// those whose chunk was freed before the demotion committed. The other
390    /// write counters count the same writes.
391    pub extent_file_writes: u64,
392    /// The subset of `extent_file_writes` run by inline enforcement passes
393    /// rather than spill threads.
394    pub extent_file_writes_inline: u64,
395    /// Bytes transferred by the `extent_file_writes` of identity-coded
396    /// extents, page-rounded by the store.
397    pub extent_file_write_bytes_identity: u64,
398    /// As `extent_file_write_bytes_identity`, for all other codecs.
399    pub extent_file_write_bytes_compressed: u64,
400    /// Reads of demoted extents.
401    pub extent_file_reads: u64,
402    /// Page-rounded bytes transferred by `extent_file_reads`.
403    pub extent_file_read_bytes: u64,
404    /// The subset of `extent_file_reads` of extents already read since their
405    /// demotion.
406    pub extent_file_repeat_reads: u64,
407    /// Demotion passes that left an extent in RAM because the store was
408    /// full or had writes disabled.
409    pub extent_file_full: u64,
410    /// Store I/O errors, after which demotion writes are disabled.
411    pub extent_file_write_errors: u64,
412    /// Bytes of file slots returned to the filesystem.
413    pub extent_file_holes_punched_bytes: u64,
414    /// Read latency histogram of `extent_file_reads`. Bucket 0 counts reads
415    /// under 32 µs, bucket `i` in `1..12` reads in `[16 µs << i, 32 µs << i)`,
416    /// and bucket 12 reads of 65.536 ms and more.
417    pub extent_file_read_latency: [u64; READ_LATENCY_BUCKETS],
418}
419
420#[derive(Debug, Default)]
421struct Counters {
422    direct_extent_inserts: AtomicU64,
423    inserts: AtomicU64,
424    spill_scheduled: AtomicU64,
425    spill_cancelled: AtomicU64,
426    slot_exhausted_fallbacks: AtomicU64,
427    oversize_payloads: AtomicU64,
428    frees: AtomicU64,
429    writes_elided: AtomicU64,
430    evictions_compress: AtomicU64,
431    evictions_cheap: AtomicU64,
432    extent_bytes_written: AtomicU64,
433    resident_bytes: AtomicU64,
434    oversize_bytes: AtomicU64,
435    warm_bytes: AtomicU64,
436    warm_reuses: AtomicU64,
437    eager_backs: AtomicU64,
438    admissions_budget: AtomicU64,
439    admissions_steal: AtomicU64,
440    admissions_denied: AtomicU64,
441    extent_resident_bytes: AtomicU64,
442    extent_unreclaimable_bytes: AtomicU64,
443    extent_pageouts: AtomicU64,
444    extent_pageout_incomplete: AtomicU64,
445    extent_demotions_elided: AtomicU64,
446    extent_file_bytes: AtomicU64,
447    extent_file_writes_inline: AtomicU64,
448    extent_file_write_bytes_identity: AtomicU64,
449    extent_file_write_bytes_compressed: AtomicU64,
450    extent_file_repeat_reads: AtomicU64,
451    extent_file_full: AtomicU64,
452}
453
454/// A buffer pool over compressed extents, backed by swap or by files per its
455/// [`ExtentBackend`]. Cheap to clone. All clones share one budget and one
456/// backing store.
457#[derive(Debug, Clone)]
458pub struct Pool(Arc<PoolInner>);
459
460/// The shared state behind every [`Pool`] handle. One per process in
461/// practice; [`Pool`] clones and chunk handles share it through an `Arc`,
462/// so it lives until the last handle and spill thread release it.
463///
464/// Lock order: a chunk's `state` mutex may be held while taking any of the
465/// leaf locks (the eviction `queue`, the `extent_queue`, the spill queue,
466/// the region slot allocators, and the file store's per-class slot
467/// allocators, which `alloc`, `free`, and hole punching take) but never the
468/// reverse. The enforcement and backing scans additionally drop the queue
469/// guard before trying a chunk's state lock, so no path holds a queue lock
470/// while waiting on chunk state. Scans only ever `try_lock` a chunk they
471/// pop, with one exception: a demotion re-locks its victim with a blocking
472/// `lock` to commit after the unlocked write. That wait holds no other
473/// chunk or queue lock. From budget enforcement it holds the `enforcing`
474/// single-flight mutex, which is only ever `try_lock`ed, so no thread waits
475/// on it. Hence `enforce_compressed_cap` must never run while its caller holds
476/// any chunk's state lock: two such callers demoting each other's chunks
477/// would deadlock on the re-lock. The admitting read's victim steal is the
478/// one place a chunk's state lock is held while probing another chunk's,
479/// and the victim is only ever `try_lock`ed, so two admitters stealing
480/// toward each other skip instead of deadlocking.
481///
482/// Reads copy out under the chunk's state lock, the same lock eviction
483/// takes, so there is no reader-side count and no reader the evictor must
484/// account for. The exception is a plain read of an extent demoted to file,
485/// which copies out after the unlock, sound per `Home::File` in `extent.rs`.
486#[derive(Debug)]
487struct PoolInner {
488    /// Resident-bytes target, enforced against evictable bytes (resident
489    /// minus heap-backed, which no eviction can reclaim). Atomic so a
490    /// running pool can be retuned in place (operator-driven budget
491    /// changes) without orphaning live handles, which share this value
492    /// through their `Arc<PoolInner>`.
493    budget_bytes: AtomicU64,
494    /// Ceiling on the pool's *total* RSS: slots (the budget) plus warm free
495    /// slots plus compressed-resident extents. The compressed tier's
496    /// capacity derives as `max(0, rss_target - budget - warm cap)`; zero
497    /// (the default) collapses the tier, paging every extent out as soon as
498    /// it is written.
499    rss_target_bytes: AtomicU64,
500    /// One region per entry of [`SIZE_CLASSES`], same order.
501    regions: Vec<Region>,
502    /// The arena backing extents. Shared with every live [`Extent`],
503    /// whose drop returns its slot.
504    extent_arena: Arc<ExtentArena>,
505    /// Where RSS-target enforcement sends arena extents: the swap device,
506    /// or a file store that demoted extents keep alive.
507    store: ExtentStore,
508    /// File mode: set when a demotion pass found the store unable to take
509    /// any extent it probed, cleared by a committed demotion or a freed file
510    /// slot, or by enqueueing an extent whose class the store can place. While
511    /// set, inline callers run no demotion pass, and each spill thread runs
512    /// one at most every [`SPILL_PARK_TIMEOUT`].
513    ///
514    /// NOTE: setting and clearing race. A pass can set the hint after a
515    /// concurrent free cleared it, and demotion then stays deferred until the
516    /// next free, commit, or enqueue of a placeable extent, or the next spill
517    /// thread retry.
518    full_hint: AtomicBool,
519    /// File mode: entries whose extent class the store refused, set aside so
520    /// that passes stop revisiting them. A pass returns a class's entries to
521    /// the front of `extent_queue` once the store can place that class.
522    parked_extents: Mutex<ParkedExtents>,
523    /// `class_bit` of every class with parked entries, so a pass skips the
524    /// parked lock when nothing is parked.
525    parked_classes: AtomicU64,
526    /// File mode: the schedule on which spill threads release warm file
527    /// store slots. See [`Pool::set_file_release`].
528    file_release: Mutex<FileRelease>,
529    /// Second-chance FIFOs of eviction candidates, one per depth band; a
530    /// chunk joins the band of its [`ChunkHints`] depth at insert and again
531    /// on re-admission. Entries for freed chunks go stale in place and are
532    /// dropped by [`PoolInner::prune_queues`].
533    ///
534    /// Two scanners walk them with different obligations, both visiting the
535    /// deepest band first. Budget enforcement is the one that ages chunks:
536    /// it spends the touched bit (second chance) and drops entries it
537    /// evicts. Eager backing ([`PoolInner::back_one`]) rotates visited
538    /// entries to the back but never spends a touched bit, so a backing
539    /// pass shuffles FIFO order without aging any chunk toward eviction.
540    queues: [Mutex<VecDeque<Weak<ChunkMeta>>>; DEPTH_BANDS],
541    /// FIFO of chunks whose extents are resident, oldest first — the
542    /// RSS-target enforcement's victim queue. Entries go stale when an
543    /// extent pages out, is dropped, or its chunk dies; visits drop them,
544    /// and [`PoolInner::prune_extent_queue`] compacts dead-chunk entries
545    /// that under-cap operation never visits.
546    extent_queue: Mutex<VecDeque<ExtentEntry>>,
547    /// Number of live size-classed chunks (whatever their residency), which
548    /// is the number of non-stale queue entries across all bands;
549    /// [`PoolInner::prune_queues`] compacts the queues against it.
550    live_chunks: AtomicU64,
551    /// Number of live chunks whose extent is currently resident, including
552    /// unreclaimable extents that deliberately hold no `extent_queue` entry
553    /// (heap-backed and retry-capped ones). It therefore upper-bounds the
554    /// queue's non-stale entries, and [`PoolInner::prune_extent_queue`]'s
555    /// compaction threshold is conservative by the unreclaimable count.
556    extent_residents: AtomicU64,
557    /// Single-flight claim for budget enforcement.
558    enforcing: Mutex<()>,
559    /// Set by an insert turned away from `enforcing`. The holder re-runs its
560    /// pass while it is set, so a caller turned away after the holder's final
561    /// counter read still has its bytes enforced rather than dropped.
562    enforce_pending: std::sync::atomic::AtomicBool,
563    counters: Counters,
564    spill: Spill,
565}
566
567/// Hand-off point between budget enforcement and spill threads. Eviction I/O
568/// (compression and the synchronous-reclaim `pageout`) runs on spill threads
569/// when enabled, keeping multi-millisecond work off the threads that trip the
570/// budget; with no spill threads, eviction runs inline on the caller.
571#[derive(Debug, Default)]
572struct Spill {
573    /// Chunks in `WriteInFlight`, awaiting a spill thread.
574    queue: Mutex<VecDeque<Arc<ChunkMeta>>>,
575    /// Parks idle spill threads. Notified when work lands in `queue`, when
576    /// eager backing turns on, and at shutdown; threads additionally wake
577    /// on a timeout so eager backing scans for write-behind work without a
578    /// dedicated wakeup per candidate.
579    cv: std::sync::Condvar,
580    /// Whether evictions are handed to spill threads. Set when threads are
581    /// first spawned; cleared to fall back to inline eviction.
582    enabled: std::sync::atomic::AtomicBool,
583    /// Whether idle spill threads eagerly compress unbacked chunks to
584    /// `BackedResident` (write-behind); see [`Pool::set_eager_backing`].
585    eager: std::sync::atomic::AtomicBool,
586    /// Number of spill threads spawned (spawn-once; later config changes
587    /// only toggle `enabled`).
588    threads: AtomicU64,
589    /// Queued plus currently-processing entries; `quiesce` waits on zero.
590    in_flight: AtomicU64,
591    /// Lifecycle: production spill threads are immortal (the pool is a
592    /// process singleton) and never set `stop`, but Miri rejects a test
593    /// binary exiting with live threads, so tests stop and join them.
594    stop: std::sync::atomic::AtomicBool,
595    handles: Mutex<Vec<std::thread::JoinHandle<()>>>,
596}
597
598/// The schedule of the file store's warm-slot release.
599#[derive(Debug)]
600struct FileRelease {
601    interval: std::time::Duration,
602    /// Bytes each class releases per interval.
603    bytes_per_class: u64,
604    /// When the next release is due. `None` when an interval too long for
605    /// `Instant` put it out of reach.
606    due: Option<std::time::Instant>,
607    /// When the next capacity refresh is due, on its own cadence so that
608    /// disabling the release does not freeze the capacity.
609    refresh_due: std::time::Instant,
610}
611
612/// Beyond this many queued or in-flight spill entries, eviction degrades to
613/// inline on the caller: bounded memory overshoot under burst beats an
614/// unbounded queue of still-resident chunks.
615pub const SPILL_IN_FLIGHT_MAX: usize = 64;
616
617/// The most extent bytes one file-mode pass demotes: a spill thread's pass
618/// while the eviction queue is empty, and an inline pass while spill threads
619/// exist. With queued evictions a spill thread's pass demotes one extent, so
620/// the thread alternates one eviction with one demotion.
621///
622/// Demotions are device writes. On a device slower than the rate at which
623/// evictions fill the compressed tier, an unbounded spill-thread pass never
624/// gets the tier back to its capacity, the eviction queue stays full, and
625/// every eviction runs inline on the threads that trip the budget. An inline
626/// pass runs on the caller's thread, often a timely worker, and an unbounded
627/// one after a lowered RSS target writes the whole overage before the worker
628/// returns. The bound is at least the largest size class, so an inline
629/// caller still demotes the extent its own eviction added. 16 MiB is about
630/// 35 ms of writes at the 475 MB/s an 8 vCPU instance-store NVMe sustained.
631const DEMOTION_PASS_BYTES: u64 = 16 << 20;
632
633/// How often a spill thread re-derives the file store's capacity from the
634/// volume's free space.
635const FILE_CAPACITY_REFRESH_INTERVAL: std::time::Duration = std::time::Duration::from_secs(1);
636
637/// How long an idle spill thread parks, and how often a spill thread retries
638/// demotion while the file store's full hint is set.
639const SPILL_PARK_TIMEOUT: std::time::Duration = std::time::Duration::from_millis(100);
640
641/// What a spill thread does with a chunk once compressed.
642#[derive(Clone, Copy, PartialEq, Eq)]
643enum SpillKind {
644    /// Budget-driven: release the slot, leaving the chunk `Evicted`.
645    Evict,
646    /// Eager write-behind: keep the slot, leaving the chunk
647    /// `BackedResident`.
648    Back,
649}
650
651/// Which kind of caller runs a [`PoolInner::enforce_compressed_cap`] pass.
652#[derive(Debug, Clone, Copy, PartialEq, Eq)]
653enum Pass {
654    /// A spill thread, between jobs.
655    Background,
656    /// Any other thread: a worker over the backstop threshold, a budget
657    /// change, or a pool without spill threads.
658    Inline,
659}
660
661/// The tier size at which callers with spill threads present enforce the
662/// compressed cap inline, and where an inline file-mode pass stops.
663fn backstop_threshold(cap: u64) -> u64 {
664    cap.saturating_mul(2)
665}
666
667/// The outcome of one [`PoolInner::demote`], as it bears on the pass.
668enum Demotion {
669    /// The extent moved to file.
670    Committed,
671    /// The entry was stale, or the chunk died mid-write.
672    Skipped,
673    /// The store cannot place the extent's class, carried in the variant, but
674    /// can place another class. The entry was parked.
675    Refused(usize),
676    /// The store can place no class or has writes disabled. The entry went to
677    /// the front of the queue, and the pass must end.
678    Stopped,
679    /// The write failed, which disabled the store's writes. The entry went to
680    /// the front of the queue, and the pass must end.
681    WriteFailed,
682}
683
684/// `extent_queue` entries set aside per extent class. See
685/// `PoolInner::parked_extents`.
686#[derive(Debug, Default)]
687struct ParkedExtents {
688    /// Indexed by file store class, each oldest first.
689    classes: Vec<VecDeque<ExtentEntry>>,
690    /// Entries across all classes.
691    len: usize,
692}
693
694/// An `extent_queue` entry.
695#[derive(Debug)]
696struct ExtentEntry {
697    chunk: Weak<ChunkMeta>,
698    /// The extent's `alloc_size` at enqueue. The file store is opened with
699    /// the arena's class ladder, so in file mode this names the extent's
700    /// file class without taking the chunk lock.
701    alloc_size: usize,
702}
703
704impl ExtentEntry {
705    fn new(meta: &Arc<ChunkMeta>, alloc_size: usize) -> ExtentEntry {
706        ExtentEntry {
707            chunk: Arc::downgrade(meta),
708            alloc_size,
709        }
710    }
711}
712
713/// The bit of `class` in a pass's refused-class mask, or 0 for a class the
714/// mask cannot represent, which the pass then probes on every visit.
715fn class_bit(class: usize) -> u64 {
716    u32::try_from(class)
717        .ok()
718        .and_then(|class| 1u64.checked_shl(class))
719        .unwrap_or(0)
720}
721
722#[derive(Debug)]
723struct ChunkMeta {
724    pool: Arc<PoolInner>,
725    /// Length in `u64` words; immutable.
726    len: usize,
727    /// Size class for slot allocations; `None` for empty chunks and payloads
728    /// beyond the largest class. Immutable: the chunk's *slot* comes and goes
729    /// with residency, but it is always drawn from this class.
730    class: Option<usize>,
731    /// The insert-time [`ChunkHints`] depth; immutable. Names the eviction
732    /// band the chunk's queue entries belong to.
733    depth: u8,
734    /// The insert-time [`ExtentCodec`]; immutable. Encodes the chunk when it
735    /// is backed and decodes its extent on reads, so it must outlive any
736    /// extent it produced, hence `'static`.
737    codec: &'static dyn ExtentCodec,
738    state: Mutex<ChunkState>,
739}
740
741#[derive(Debug)]
742struct ChunkState {
743    residency: Residency,
744    /// Second-chance bit, set on read and cleared (in lieu of eviction) when
745    /// the budget enforcer first visits the chunk.
746    touched: bool,
747    /// Set when the owning handle is dropped, so a queue entry upgraded
748    /// concurrently with the free cannot touch a recycled slot.
749    freed: bool,
750    /// The chunk's slot index within its class's region, held exactly while
751    /// the chunk occupies pool memory (the resident states and
752    /// `WriteInFlight`). Eviction returns the slot to the region free list.
753    /// Reads copy the slot out under the state lock; no pointer into the
754    /// slot outlives the lock under which it was formed.
755    slot: Option<u32>,
756    /// The resident body of a chunk inserted by [`Pool::insert_typed`], in
757    /// place of a slot. A resident chunk holds at most one of `slot` and
758    /// `typed`; eviction drops whichever it holds.
759    typed: Option<Box<dyn TypedBody>>,
760    /// The backing copy; present exactly in the `BackedResident` and
761    /// `Evicted` states, and in a freed chunk whose extent a demoter still
762    /// owns.
763    extent: Option<Extent>,
764    /// The payload of an `Oversize` chunk.
765    oversize: Option<Vec<u64>>,
766}
767
768impl ChunkMeta {
769    /// A fresh chunk in its insert-time state.
770    fn new(
771        pool: &Arc<PoolInner>,
772        len: usize,
773        class: Option<usize>,
774        depth: u8,
775        codec: &'static dyn ExtentCodec,
776        residency: Residency,
777        slot: Option<u32>,
778        oversize: Option<Vec<u64>>,
779    ) -> ChunkMeta {
780        ChunkMeta {
781            pool: Arc::clone(pool),
782            len,
783            class,
784            depth,
785            codec,
786            state: Mutex::new(ChunkState {
787                residency,
788                touched: false,
789                freed: false,
790                slot,
791                typed: None,
792                extent: None,
793                oversize,
794            }),
795        }
796    }
797
798    fn len_bytes(&self) -> usize {
799        self.len * WORD_BYTES
800    }
801
802    /// Locks the chunk's state.
803    fn state(&self) -> MutexGuard<'_, ChunkState> {
804        self.state.lock().expect("chunk state poisoned")
805    }
806}
807
808/// Handle to one immutable chunk in a [`Pool`]. Dropping the handle frees the
809/// chunk: the slot (if resident) returns to the region free list with its
810/// physical pages released, and the extent (if any) is deallocated,
811/// discarding any swapped copy for free.
812#[derive(Debug)]
813pub struct ChunkHandle {
814    meta: Arc<ChunkMeta>,
815}
816
817impl Pool {
818    /// Creates a pool over swap-backed extents, reserving one virtual
819    /// region per size class. The pool starts with an unlimited budget, so
820    /// nothing is evicted until [`Pool::set_budget`] tunes it.
821    pub fn new() -> std::io::Result<Pool> {
822        Pool::with_backend(ExtentBackend::Swap)
823    }
824
825    /// As [`Pool::new`], over `backend`. A file backend that cannot open
826    /// returns the error, and the caller decides whether to fall back.
827    pub fn with_backend(backend: ExtentBackend) -> std::io::Result<Pool> {
828        Pool::with_backend_and_class_capacity(backend, CLASS_CAPACITY_BYTES)
829    }
830
831    /// The store the pool runs on.
832    pub fn backend_kind(&self) -> BackendKind {
833        match &self.0.store {
834            ExtentStore::Swap => BackendKind::Swap,
835            ExtentStore::File(store) => match store.io_mode() {
836                IoMode::Direct => BackendKind::FileDirect,
837                IoMode::Buffered => BackendKind::FileBuffered,
838            },
839        }
840    }
841
842    /// As [`Pool::with_backend`], with a caller-chosen virtual reservation
843    /// per size class.
844    fn with_backend_and_class_capacity(
845        backend: ExtentBackend,
846        class_capacity_bytes: usize,
847    ) -> std::io::Result<Pool> {
848        let extent_arena = Arc::new(ExtentArena::new(class_capacity_bytes)?);
849        // Demotion moves an arena extent to the file slot of the same class
850        // index, so the store must use the arena's ladder.
851        let store = match backend {
852            ExtentBackend::Swap => ExtentStore::Swap,
853            ExtentBackend::File {
854                dir,
855                capacity_bytes,
856            } => ExtentStore::File(Arc::new(FileStore::open(
857                &dir,
858                capacity_bytes,
859                extent_arena.classes(),
860            )?)),
861        };
862        let regions = SIZE_CLASSES
863            .iter()
864            .map(|&class_size| Region::new(class_size, class_capacity_bytes))
865            .collect::<std::io::Result<Vec<_>>>()?;
866        Ok(Pool(Arc::new(PoolInner {
867            budget_bytes: AtomicU64::new(u64::MAX),
868            rss_target_bytes: AtomicU64::new(0),
869            regions,
870            extent_arena,
871            store,
872            full_hint: AtomicBool::new(false),
873            parked_extents: Mutex::new(ParkedExtents::default()),
874            parked_classes: AtomicU64::new(0),
875            file_release: Mutex::new(FileRelease {
876                interval: std::time::Duration::ZERO,
877                bytes_per_class: 0,
878                due: Some(std::time::Instant::now()),
879                refresh_due: std::time::Instant::now(),
880            }),
881            queues: std::array::from_fn(|_| Mutex::new(VecDeque::new())),
882            extent_queue: Mutex::new(VecDeque::new()),
883            live_chunks: AtomicU64::new(0),
884            extent_residents: AtomicU64::new(0),
885            enforcing: Mutex::new(()),
886            enforce_pending: std::sync::atomic::AtomicBool::new(false),
887            counters: Counters::default(),
888            spill: Spill::default(),
889        })))
890    }
891
892    /// Allocates a chunk of `len` words and fills it in place: `fill`
893    /// receives `len` words of unspecified contents and must overwrite all
894    /// of them, so serialization writes its single copy straight into pool
895    /// memory. The returned handle starts `UnbackedResident`, or `Evicted`
896    /// when resident admission is full even after enforcement, in which case
897    /// the payload is compressed to an extent on the calling thread before
898    /// this returns. A zero `len` returns a length-0 handle
899    /// holding no slot; payloads beyond the largest size class fall back to
900    /// a plain heap allocation, always resident, a prototype limitation.
901    /// `hints` steer eviction and write-behind policy; callers without
902    /// placement knowledge pass the default. `codec` is the chunk's
903    /// [`ExtentCodec`], fixed for its lifetime: the pool invokes it whenever
904    /// the chunk moves across the extent boundary, and takes no interest in
905    /// the stored form it produces.
906    ///
907    /// Relies on abort-on-panic: a panic in `fill` that was caught would
908    /// leak the slot and its resident-bytes accounting. All hosting
909    /// binaries abort via `mz_ore::panic::install_enhanced_handler`, and
910    /// pool consumers are dataflow operators, never code hosted under a
911    /// `catch_unwind` boundary the way the optimizer is.
912    pub fn insert_with(
913        &self,
914        len: usize,
915        hints: ChunkHints,
916        codec: &'static dyn ExtentCodec,
917        fill: impl FnOnce(&mut [u64]),
918    ) -> ChunkHandle {
919        let inner = &self.0;
920        inner.counters.inserts.fetch_add(1, Ordering::Relaxed);
921        let len_bytes = len * WORD_BYTES;
922        if len == 0 {
923            fill(&mut []);
924            let meta = ChunkMeta::new(
925                inner,
926                0,
927                None,
928                hints.depth,
929                codec,
930                Residency::UnbackedResident,
931                None,
932                None,
933            );
934            return ChunkHandle {
935                meta: Arc::new(meta),
936            };
937        }
938        let class = region::size_class_for(len_bytes);
939        if class.is_none() {
940            inner
941                .counters
942                .oversize_payloads
943                .fetch_add(1, Ordering::Relaxed);
944        }
945        if class.is_some() {
946            if !inner.reserve_insert(len_bytes) {
947                inner.enforce_budget();
948                if !inner.reserve_insert(len_bytes) {
949                    return self.insert_extent(len, class, hints, codec, fill);
950                }
951            }
952        } else {
953            inner
954                .counters
955                .resident_bytes
956                .fetch_add(u64::cast_from(len_bytes), Ordering::Relaxed);
957        }
958        // A class with no free slot degrades to the heap path below: an
959        // unpageable chunk beats a dead replica.
960        let slot = class.and_then(|class| inner.alloc_slot(class, len_bytes));
961        let meta = match (class, slot) {
962            (Some(class), Some(slot)) => {
963                let region = &inner.regions[class];
964                // SAFETY: the freshly allocated slot is at least `len_bytes`
965                // long (the class fits the payload) and is exclusively owned
966                // by this not-yet-shared chunk, so the mutable borrow is
967                // unique; region memory is mapped and writable, and `u64` has
968                // no validity requirements beyond size, so exposing the
969                // unspecified prior contents through `&mut [u64]` is sound.
970                let dst = unsafe {
971                    std::slice::from_raw_parts_mut(region.slot_ptr(slot).cast::<u64>(), len)
972                };
973                // The fill contract (overwrite all `len` words) is
974                // discipline-only. Poison in debug builds so an
975                // under-writing fill reads back as deterministic garbage
976                // instead of a previous occupant's bytes, which the heap
977                // path's zero fill would otherwise mask in tests.
978                #[cfg(debug_assertions)]
979                dst.fill(u64::from_ne_bytes([0xDE; 8]));
980                fill(dst);
981                ChunkMeta::new(
982                    inner,
983                    len,
984                    Some(class),
985                    hints.depth,
986                    codec,
987                    Residency::UnbackedResident,
988                    Some(slot),
989                    None,
990                )
991            }
992            _ => {
993                let mut payload = vec![0u64; len];
994                fill(&mut payload);
995                inner
996                    .counters
997                    .oversize_bytes
998                    .fetch_add(u64::cast_from(len_bytes), Ordering::Relaxed);
999                ChunkMeta::new(
1000                    inner,
1001                    len,
1002                    None,
1003                    hints.depth,
1004                    codec,
1005                    Residency::Oversize,
1006                    None,
1007                    Some(payload),
1008                )
1009            }
1010        };
1011        let meta = Arc::new(meta);
1012        if meta.class.is_some() {
1013            inner.live_chunks.fetch_add(1, Ordering::Relaxed);
1014            inner
1015                .queue(band(meta.depth))
1016                .push_back(Arc::downgrade(&meta));
1017        }
1018        inner.enforce_budget();
1019        ChunkHandle { meta }
1020    }
1021
1022    /// Inserts a chunk in its owner's in-memory form (see [`TypedBody`]),
1023    /// charged against the budget at its serialized length. Admission
1024    /// follows [`Pool::insert_with`]: a chunk the budget denies even after
1025    /// enforcement is serialized and compressed on the calling thread, and a
1026    /// body with no size class to evict into is serialized onto the heap.
1027    pub fn insert_typed(
1028        &self,
1029        body: Box<dyn TypedBody>,
1030        hints: ChunkHints,
1031        codec: &'static dyn ExtentCodec,
1032    ) -> ChunkHandle {
1033        let len = body.words();
1034        let len_bytes = len * WORD_BYTES;
1035        let class = region::size_class_for(len_bytes).filter(|_| len > 0);
1036        let inner = &self.0;
1037        if class.is_none() {
1038            return self.insert_with(len, hints, codec, |dst| body.serialize_into(dst));
1039        }
1040        if !inner.reserve_insert(len_bytes) {
1041            inner.enforce_budget();
1042            if !inner.reserve_insert(len_bytes) {
1043                inner.counters.inserts.fetch_add(1, Ordering::Relaxed);
1044                return self
1045                    .insert_extent(len, class, hints, codec, |dst| body.serialize_into(dst));
1046            }
1047        }
1048        inner.counters.inserts.fetch_add(1, Ordering::Relaxed);
1049        let meta = ChunkMeta::new(
1050            inner,
1051            len,
1052            class,
1053            hints.depth,
1054            codec,
1055            Residency::UnbackedResident,
1056            None,
1057            None,
1058        );
1059        meta.state().typed = Some(body);
1060        let meta = Arc::new(meta);
1061        inner.live_chunks.fetch_add(1, Ordering::Relaxed);
1062        inner
1063            .queue(band(meta.depth))
1064            .push_back(Arc::downgrade(&meta));
1065        inner.enforce_budget();
1066        ChunkHandle { meta }
1067    }
1068
1069    /// Inserts a chunk directly as a compressed extent, with no resident slot.
1070    ///
1071    /// The fallback when the budget denies an insertion even after
1072    /// enforcement. The payload is filled into a temporary buffer and
1073    /// compressed on the calling thread, and the chunk starts evicted, so a
1074    /// denied insertion adds no resident bytes while the enforcer is busy.
1075    /// The temporary buffer, up to one size class, is heap memory outside the
1076    /// pool's accounting. A chunk inserted here pays for compression even if
1077    /// it dies young, so under sustained pressure `writes_elided` undercounts
1078    /// the writes a resident insertion would have avoided.
1079    fn insert_extent(
1080        &self,
1081        len: usize,
1082        class: Option<usize>,
1083        hints: ChunkHints,
1084        codec: &'static dyn ExtentCodec,
1085        fill: impl FnOnce(&mut [u64]),
1086    ) -> ChunkHandle {
1087        // Compress synchronously to keep denied insertions from queuing
1088        // uncompressed payloads behind an occupied enforcer.
1089        let mut words = vec![0; len];
1090        // Poisoned for the same reason as the slot path in `insert_with`.
1091        #[cfg(debug_assertions)]
1092        words.fill(u64::from_ne_bytes([0xDE; 8]));
1093        fill(&mut words);
1094        let inner = &self.0;
1095        let extent = Extent::write(&inner.extent_arena, &words, codec, Scratch::Shrink);
1096        drop(words);
1097        let meta = Arc::new(ChunkMeta::new(
1098            inner,
1099            len,
1100            class,
1101            hints.depth,
1102            codec,
1103            Residency::Evicted,
1104            None,
1105            None,
1106        ));
1107        inner.live_chunks.fetch_add(1, Ordering::Relaxed);
1108        inner
1109            .counters
1110            .direct_extent_inserts
1111            .fetch_add(1, Ordering::Relaxed);
1112        {
1113            let mut state = meta.state();
1114            inner.commit_extent(&meta, &mut state, extent);
1115        }
1116        inner.enforce_or_defer_compressed_cap();
1117        ChunkHandle { meta }
1118    }
1119
1120    /// Snapshot of the pool's counters.
1121    pub fn stats(&self) -> PoolStats {
1122        let c = &self.0.counters;
1123        let (file, file_capacity) = match &self.0.store {
1124            ExtentStore::Swap => (file::FileStoreStats::default(), 0),
1125            ExtentStore::File(store) => (store.stats(), store.capacity_bytes()),
1126        };
1127        PoolStats {
1128            extent_demotions_elided: c.extent_demotions_elided.load(Ordering::Relaxed),
1129            extent_file_bytes: c.extent_file_bytes.load(Ordering::Relaxed),
1130            extent_file_allocated_bytes: file.allocated_bytes,
1131            extent_file_capacity_bytes: file_capacity,
1132            extent_file_writes: file.writes,
1133            extent_file_writes_inline: c.extent_file_writes_inline.load(Ordering::Relaxed),
1134            extent_file_write_bytes_identity: c
1135                .extent_file_write_bytes_identity
1136                .load(Ordering::Relaxed),
1137            extent_file_write_bytes_compressed: c
1138                .extent_file_write_bytes_compressed
1139                .load(Ordering::Relaxed),
1140            extent_file_reads: file.reads,
1141            extent_file_read_bytes: file.read_bytes,
1142            extent_file_repeat_reads: c.extent_file_repeat_reads.load(Ordering::Relaxed),
1143            extent_file_full: c.extent_file_full.load(Ordering::Relaxed),
1144            extent_file_write_errors: file.write_errors,
1145            extent_file_holes_punched_bytes: file.holes_punched_bytes,
1146            extent_file_read_latency: file.read_latency_buckets,
1147            inserts: c.inserts.load(Ordering::Relaxed),
1148            direct_extent_inserts: c.direct_extent_inserts.load(Ordering::Relaxed),
1149            frees: c.frees.load(Ordering::Relaxed),
1150            writes_elided: c.writes_elided.load(Ordering::Relaxed),
1151            evictions_compress: c.evictions_compress.load(Ordering::Relaxed),
1152            evictions_cheap: c.evictions_cheap.load(Ordering::Relaxed),
1153            extent_bytes_written: c.extent_bytes_written.load(Ordering::Relaxed),
1154            resident_bytes: c.resident_bytes.load(Ordering::Relaxed),
1155            oversize_bytes: c.oversize_bytes.load(Ordering::Relaxed),
1156            warm_bytes: c.warm_bytes.load(Ordering::Relaxed),
1157            warm_reuses: c.warm_reuses.load(Ordering::Relaxed),
1158            eager_backs: c.eager_backs.load(Ordering::Relaxed),
1159            admissions_budget: c.admissions_budget.load(Ordering::Relaxed),
1160            admissions_steal: c.admissions_steal.load(Ordering::Relaxed),
1161            admissions_denied: c.admissions_denied.load(Ordering::Relaxed),
1162            extent_resident_bytes: c.extent_resident_bytes.load(Ordering::Relaxed),
1163            extent_unreclaimable_bytes: c.extent_unreclaimable_bytes.load(Ordering::Relaxed),
1164            extent_pageouts: c.extent_pageouts.load(Ordering::Relaxed),
1165            extent_pageout_incomplete: c.extent_pageout_incomplete.load(Ordering::Relaxed),
1166            extent_arena_fallbacks: self.0.extent_arena.fallbacks(),
1167            spill_scheduled: c.spill_scheduled.load(Ordering::Relaxed),
1168            spill_cancelled: c.spill_cancelled.load(Ordering::Relaxed),
1169            spill_in_flight: self.0.spill.in_flight.load(Ordering::Relaxed),
1170            slot_exhausted_fallbacks: c.slot_exhausted_fallbacks.load(Ordering::Relaxed),
1171            oversize_payloads: c.oversize_payloads.load(Ordering::Relaxed),
1172            live_chunks: self.0.live_chunks.load(Ordering::Relaxed),
1173        }
1174    }
1175
1176    /// Enables or disables off-worker eviction I/O. The first call with
1177    /// `threads > 0` spawns that many spill threads (spawn-once: later calls
1178    /// only toggle participation); `threads == 0` falls back to inline
1179    /// eviction on the caller for subsequent victims, letting any queued
1180    /// work drain.
1181    pub fn set_spill_threads(&self, threads: usize) {
1182        if threads == 0 {
1183            self.0.spill.enabled.store(false, Ordering::Relaxed);
1184            return;
1185        }
1186        let spawned = self.0.spill.threads.load(Ordering::Relaxed);
1187        if spawned == 0 {
1188            let to_spawn = u64::cast_from(threads);
1189            if self
1190                .0
1191                .spill
1192                .threads
1193                .compare_exchange(0, to_spawn, Ordering::Relaxed, Ordering::Relaxed)
1194                .is_ok()
1195            {
1196                for i in 0..threads {
1197                    let inner = Arc::clone(&self.0);
1198                    let handle = std::thread::Builder::new()
1199                        .name(format!("pool-spill-{i}"))
1200                        .spawn(move || inner.spill_worker())
1201                        .expect("spawn pool spill thread");
1202                    self.0
1203                        .spill
1204                        .handles
1205                        .lock()
1206                        .expect("spill handles poisoned")
1207                        .push(handle);
1208                }
1209            }
1210        }
1211        self.0.spill.enabled.store(true, Ordering::Relaxed);
1212    }
1213
1214    /// Enables or disables eager backing: when on, idle spill threads
1215    /// compress unbacked chunks to `BackedResident` ahead of pressure, so
1216    /// budget-driven eviction becomes a pure page release. Costs CPU on
1217    /// chunks that die before eviction would have reached them; pays at
1218    /// every pressure event. Only meaningful with spill threads spawned.
1219    pub fn set_eager_backing(&self, eager: bool) {
1220        self.0.spill.eager.store(eager, Ordering::Relaxed);
1221        if eager {
1222            self.0.spill.cv.notify_all();
1223        }
1224    }
1225
1226    /// Retunes the file store's warm-slot release: every `interval`, a spill
1227    /// thread punches up to `bytes_per_class` of each extent class's free
1228    /// slots that still hold blocks, so the store's disk usage follows its
1229    /// live extents. A zero interval or zero bytes disables the release. No
1230    /// effect on a swap-backed pool or one without spill threads.
1231    ///
1232    /// The store's capacity follows the volume's free space independently of
1233    /// the release, re-derived by the spill threads every second. Without
1234    /// spill threads it changes only when the store hits `ENOSPC`.
1235    pub fn set_file_release(&self, interval: std::time::Duration, bytes_per_class: u64) {
1236        let mut release = self.0.file_release();
1237        release.interval = interval;
1238        release.bytes_per_class = bytes_per_class;
1239        // A shortened interval takes effect now rather than after the
1240        // current one. Taking the minimum keeps the per-tick reapplication
1241        // of an unchanged config from postponing the release.
1242        let next = std::time::Instant::now().checked_add(interval);
1243        release.due = release.due.into_iter().chain(next).min();
1244    }
1245
1246    /// Retunes the resident-bytes budget in place and enforces it. Live
1247    /// handles share the new value immediately through their `Arc<PoolInner>`;
1248    /// a shrink takes effect by evicting on this call, a grow simply leaves
1249    /// more headroom for future inserts.
1250    pub fn set_budget(&self, budget_bytes: usize) {
1251        let new = u64::cast_from(budget_bytes);
1252        let prev = self.0.budget_bytes.swap(new, Ordering::Relaxed);
1253        // Config application calls this per worker per tick; only a change
1254        // warrants an enforcement pass (a grow needs none, and inserts
1255        // enforce continuously anyway).
1256        if new < prev {
1257            self.0.trim_warm_pool();
1258            self.0.enforce_budget();
1259        }
1260    }
1261
1262    /// Retunes the ceiling on the pool's total RSS — slots plus warm slots
1263    /// plus compressed-resident extents. The compressed tier's capacity is
1264    /// the gap above the budget and warm cap; zero (the default) collapses
1265    /// the tier, paging extents out as soon as they are written. A shrink
1266    /// starts paging out the oldest extents on this call.
1267    pub fn set_rss_target(&self, target_bytes: usize) {
1268        let new = u64::cast_from(target_bytes);
1269        let prev = self.0.rss_target_bytes.swap(new, Ordering::Relaxed);
1270        if new < prev {
1271            self.0.enforce_compressed_cap(Pass::Inline);
1272            // An inline file-mode pass stops at the backstop threshold or
1273            // after `DEMOTION_PASS_BYTES`, and the spill threads demote the
1274            // rest.
1275            self.0.spill.cv.notify_all();
1276        }
1277    }
1278
1279    /// The capacity of the compressed tier in bytes: the RSS target less the
1280    /// budget, the insert slack in use, and the warm-slot cap.
1281    pub fn compressed_cap(&self) -> u64 {
1282        self.0.compressed_cap()
1283    }
1284
1285    /// Whether the compressed tier is above [`Pool::compressed_cap`], counted
1286    /// as the backend's enforcement counts it. False once enforcement and
1287    /// the spill threads have caught up.
1288    pub fn compressed_tier_above_cap(&self) -> bool {
1289        self.0.compressed_tier_above(self.0.compressed_cap())
1290    }
1291}
1292
1293impl PoolInner {
1294    /// Locks the eviction queue of one depth band.
1295    fn queue(&self, band: usize) -> MutexGuard<'_, VecDeque<Weak<ChunkMeta>>> {
1296        self.queues[band].lock().expect("pool queue poisoned")
1297    }
1298
1299    /// Locks the resident-extent queue.
1300    fn extent_queue(&self) -> MutexGuard<'_, VecDeque<ExtentEntry>> {
1301        self.extent_queue.lock().expect("extent queue poisoned")
1302    }
1303
1304    /// Locks the parked resident-extent entries.
1305    fn parked_extents(&self) -> MutexGuard<'_, ParkedExtents> {
1306        self.parked_extents.lock().expect("parked extents poisoned")
1307    }
1308
1309    /// Locks the spill hand-off queue.
1310    fn spill_queue(&self) -> MutexGuard<'_, VecDeque<Arc<ChunkMeta>>> {
1311        self.spill.queue.lock().expect("spill queue poisoned")
1312    }
1313
1314    /// The region behind a slotted chunk's size class.
1315    fn region_of(&self, meta: &ChunkMeta) -> &Region {
1316        &self.regions[meta.class.expect("slotted chunk has a class")]
1317    }
1318
1319    /// Borrows the payload of a slotted chunk.
1320    ///
1321    /// # Safety
1322    ///
1323    /// `slot` must be `meta`'s slot, its contents must be initialized (they
1324    /// are from insert onward), and nothing may write the slot while the
1325    /// borrow lives.
1326    unsafe fn slot_data(&self, meta: &ChunkMeta, slot: u32) -> &[u64] {
1327        let ptr = self
1328            .region_of(meta)
1329            .slot_ptr(slot)
1330            .cast_const()
1331            .cast::<u64>();
1332        // SAFETY: per the function contract; `meta.len` words fit the class
1333        // by construction.
1334        unsafe { std::slice::from_raw_parts(ptr, meta.len) }
1335    }
1336
1337    /// Records a freshly written extent under the chunk's state lock: the
1338    /// compressed-bytes counter, the compressed-tier accounting, and the
1339    /// state's extent field.
1340    fn commit_extent(&self, meta: &Arc<ChunkMeta>, state: &mut ChunkState, extent: Extent) {
1341        self.counters
1342            .extent_bytes_written
1343            .fetch_add(u64::cast_from(extent.comp_len()), Ordering::Relaxed);
1344        // A heap-fallback extent is born permanently capped and counts as
1345        // unreclaimable from the start.
1346        self.note_extent_resident(meta, extent.alloc_size(), !extent.pageout_capped());
1347        state.extent = Some(extent);
1348    }
1349
1350    /// Drops queue entries whose chunk has been freed, detected by their
1351    /// `Weak` no longer holding a live chunk. Each band compacts only when
1352    /// its stale entries outnumber all live chunks (plus a small floor), so
1353    /// the cost amortizes to a constant per insert and the total queue
1354    /// length stays proportional to the number of live slotted chunks even
1355    /// when the pool never comes under budget pressure.
1356    fn prune_queues(&self) {
1357        let live = usize::cast_from(self.live_chunks.load(Ordering::Relaxed));
1358        for band in 0..DEPTH_BANDS {
1359            let mut queue = self.queue(band);
1360            if queue.len() > 2 * live + 16 {
1361                queue.retain(|weak| weak.strong_count() > 0);
1362            }
1363        }
1364    }
1365
1366    /// Reserve insertion bytes before populating a slot. Enforcement can
1367    /// lag by [`insert_slack`] of the budget, or one payload for small
1368    /// budgets. Read admissions use the budget itself and cannot consume
1369    /// this slack.
1370    fn reserve_insert(&self, len_bytes: usize) -> bool {
1371        let len = u64::cast_from(len_bytes);
1372        self.counters
1373            .resident_bytes
1374            .try_update(Ordering::Relaxed, Ordering::Relaxed, |cur| {
1375                let budget = self.budget_bytes.load(Ordering::Relaxed);
1376                let ceiling = budget.saturating_add(insert_slack(budget).max(len));
1377                let next = cur.checked_add(len)?;
1378                let oversize = self.counters.oversize_bytes.load(Ordering::Relaxed);
1379                (next.saturating_sub(oversize) <= ceiling).then_some(next)
1380            })
1381            .is_ok()
1382    }
1383
1384    fn enforce_budget(&self) {
1385        // Single-flight: enforcement runs synchronously on whichever thread
1386        // trips it (every insert), and concurrent passes would
1387        // convoy on the queue mutex doing redundant scans of the same
1388        // candidates. One pass at a time reaches the budget just as well;
1389        // skipped callers hand their bytes to the in-progress pass through
1390        // `enforce_pending`. A poisoned claim means a prior pass panicked.
1391        // Recover and keep enforcing rather than silently disabling the
1392        // budget for the process's lifetime.
1393        let guard = match self.enforcing.try_lock() {
1394            Ok(guard) => guard,
1395            Err(std::sync::TryLockError::WouldBlock) => {
1396                // Release pairs with the holder's Acquire: any `resident_bytes`
1397                // bump this caller made must be visible to the re-read.
1398                self.enforce_pending.store(true, Ordering::Release);
1399                return;
1400            }
1401            Err(std::sync::TryLockError::Poisoned(poisoned)) => poisoned.into_inner(),
1402        };
1403        loop {
1404            self.enforce_budget_inner();
1405            #[cfg(test)]
1406            testing::run_enforce_budget_hook();
1407            // A caller turned away since this pass's counter reads may have
1408            // left bytes unenforced; re-run rather than drop them. The Acquire
1409            // pairs with the turned-away Release so the re-read sees the bump,
1410            // and this swap is the only place the flag is cleared, so no set
1411            // can be lost.
1412            //
1413            // NOTE: a caller turned away between this swap and `drop(guard)`
1414            // sets the flag but finds no re-reader. That residual window is a
1415            // few instructions wide, versus the whole pass before.
1416            if !self.enforce_pending.swap(false, Ordering::Acquire) {
1417                break;
1418            }
1419        }
1420        drop(guard);
1421        // Inline evictions above may have grown the compressed tier.
1422        self.enforce_or_defer_compressed_cap();
1423    }
1424
1425    /// Bytes budget enforcement can actually reclaim: resident bytes minus
1426    /// heap-backed (oversize and class-exhaustion) chunks, which hold no
1427    /// slot and can never be evicted. Enforcing against raw resident bytes
1428    /// would, once unevictable bytes alone exceed the budget, compress
1429    /// every slotted chunk on arrival forever.
1430    fn evictable_bytes(&self) -> u64 {
1431        self.counters
1432            .resident_bytes
1433            .load(Ordering::Relaxed)
1434            .saturating_sub(self.counters.oversize_bytes.load(Ordering::Relaxed))
1435    }
1436
1437    fn enforce_budget_inner(&self) {
1438        self.prune_queues();
1439        // Deepest band first: deep chunks are the coldest, and once eager
1440        // backing has visited them (same order) their eviction is a pure
1441        // page release. The youngest band is reached only when the deeper
1442        // bands cannot satisfy the budget, keeping young data's
1443        // die-before-write chance longest.
1444        //
1445        // The pass evicts at most the overage it saw on entry. Inserts that
1446        // land during the pass are left to the pass their own enforcement
1447        // request reruns (see `enforce_pending`), so one call does not chase
1448        // every other thread's inserts.
1449        let mut quota = self
1450            .evictable_bytes()
1451            .saturating_sub(self.budget_bytes.load(Ordering::Relaxed));
1452        for band in (0..DEPTH_BANDS).rev() {
1453            if quota == 0 || self.evictable_bytes() <= self.budget_bytes.load(Ordering::Relaxed) {
1454                return;
1455            }
1456            self.enforce_budget_band(band, &mut quota);
1457        }
1458    }
1459
1460    fn enforce_budget_band(&self, band: usize, quota: &mut u64) {
1461        // The queue holds resident chunks only (entries for evicted chunks
1462        // are dropped on visit and never re-added), so a full pass is
1463        // proportional to the resident set. Visit each queued chunk at most
1464        // twice per call: a first visit may only clear the second-chance
1465        // bit, so a second is needed before an over-budget call is
1466        // guaranteed to evict every chunk it saw. The bound keeps contended
1467        // and in-flight entries from spinning this loop forever.
1468        let mut remaining = self.queue(band).len().saturating_mul(2);
1469        while remaining > 0
1470            && *quota > 0
1471            && self.evictable_bytes() > self.budget_bytes.load(Ordering::Relaxed)
1472        {
1473            remaining -= 1;
1474            let popped = self.queue(band).pop_front();
1475            let Some(weak) = popped else {
1476                break;
1477            };
1478            let Some(meta) = weak.upgrade() else {
1479                continue;
1480            };
1481            let mut inline = false;
1482            let requeue = {
1483                // `try_lock`: a chunk mid-eviction or mid-read holds its
1484                // lock for milliseconds; skipping it beats convoying every
1485                // budget enforcer in the process behind one chunk's I/O.
1486                let Ok(mut state) = meta.state.try_lock() else {
1487                    self.queue(band).push_back(weak);
1488                    continue;
1489                };
1490                if state.freed {
1491                    false
1492                } else if matches!(state.residency, Residency::Evicted | Residency::Oversize) {
1493                    // Nothing to evict: drop the entry. A chunk re-enters
1494                    // the queue only when it becomes resident again (insert
1495                    // or re-admission), so the queue stays proportional to
1496                    // the resident set rather than accumulating every chunk
1497                    // ever evicted.
1498                    false
1499                } else if state.residency == Residency::WriteInFlight {
1500                    // Already handed off, by this pass or an earlier one,
1501                    // and charged then. Charging it again would end the pass
1502                    // before it reaches the overage it saw.
1503                    true
1504                } else if state.touched {
1505                    state.touched = false;
1506                    true
1507                } else if self.spill_handoff(&meta, &mut state) {
1508                    *quota = quota.saturating_sub(u64::cast_from(meta.len_bytes()));
1509                    // Stays queued while in flight; once the spill commits to
1510                    // `Evicted`, the next visit drops the entry.
1511                    true
1512                } else {
1513                    self.evict_locked(&meta, &mut state);
1514                    let evicted = state.residency == Residency::Evicted;
1515                    if evicted {
1516                        *quota = quota.saturating_sub(u64::cast_from(meta.len_bytes()));
1517                        inline = true;
1518                    }
1519                    !evicted
1520                }
1521            };
1522            if requeue {
1523                self.queue(band).push_back(weak);
1524            }
1525            // An inline eviction grows the compressed tier by one extent. A
1526            // single pass can evict a whole budget's worth when other threads
1527            // keep inserting, so the tier is trimmed between evictions, not
1528            // after the pass. Enforcement takes chunk locks itself, so this
1529            // runs after the victim's lock is released.
1530            if inline {
1531                self.enforce_or_defer_compressed_cap();
1532            }
1533        }
1534    }
1535
1536    fn evict_locked(&self, meta: &Arc<ChunkMeta>, state: &mut ChunkState) {
1537        if state.freed || (state.slot.is_none() && state.typed.is_none()) {
1538            return;
1539        }
1540        match state.residency {
1541            Residency::UnbackedResident => {
1542                // Inline eviction runs on whichever thread tripped the
1543                // budget, so the compression scratch must not stay parked
1544                // on it.
1545                let extent = match (state.slot, state.typed.as_deref()) {
1546                    (Some(slot), _) => {
1547                        // SAFETY: the slot belongs to this live chunk and the
1548                        // state lock is held, so nothing else touches the
1549                        // slot while this borrow is live (reads copy out
1550                        // under the same lock).
1551                        let data = unsafe { self.slot_data(meta, slot) };
1552                        Extent::write(&self.extent_arena, data, meta.codec, Scratch::Shrink)
1553                    }
1554                    (None, Some(body)) => {
1555                        let data = serialize_typed(body);
1556                        Extent::write(&self.extent_arena, &data, meta.codec, Scratch::Shrink)
1557                    }
1558                    (None, None) => unreachable!("checked above"),
1559                };
1560                self.counters
1561                    .evictions_compress
1562                    .fetch_add(1, Ordering::Relaxed);
1563                self.commit_extent(meta, state, extent);
1564            }
1565            Residency::BackedResident => {
1566                self.counters
1567                    .evictions_cheap
1568                    .fetch_add(1, Ordering::Relaxed);
1569            }
1570            Residency::WriteInFlight | Residency::Evicted | Residency::Oversize => return,
1571        }
1572        // `release_slot`'s precondition holds: the state lock is held and
1573        // `!freed` was checked above under it.
1574        self.release_resident(meta, state);
1575        state.residency = Residency::Evicted;
1576    }
1577
1578    /// Whether the next eviction should be handed to spill threads: enabled,
1579    /// and the queue is below the backpressure bound (beyond it, callers
1580    /// evict inline rather than growing an unbounded queue of still-resident
1581    /// chunks).
1582    fn spill_eligible(&self) -> bool {
1583        self.spill.enabled.load(Ordering::Relaxed)
1584            && usize::cast_from(self.spill.in_flight.load(Ordering::Relaxed)) < SPILL_IN_FLIGHT_MAX
1585    }
1586
1587    /// Hands a `WriteInFlight` chunk to the spill threads.
1588    fn spill_schedule(&self, meta: Arc<ChunkMeta>) {
1589        self.counters
1590            .spill_scheduled
1591            .fetch_add(1, Ordering::Relaxed);
1592        self.spill.in_flight.fetch_add(1, Ordering::Relaxed);
1593        self.spill_queue().push_back(meta);
1594        self.spill.cv.notify_one();
1595    }
1596
1597    /// Spill-thread main loop. The thread owns an `Arc<PoolInner>`, so the
1598    /// pool (a process-wide singleton in production) lives as long as its
1599    /// threads. Queued (budget-driven) evictions take priority; with eager
1600    /// backing enabled, idle threads compress unbacked chunks to
1601    /// `BackedResident` instead of parking, and park with a timeout once
1602    /// everything reachable is backed.
1603    fn spill_worker(self: Arc<Self>) {
1604        let mut hinted_retry = std::time::Instant::now();
1605        loop {
1606            if self.spill.stop.load(Ordering::Relaxed) {
1607                return;
1608            }
1609            self.spill_trim(&mut hinted_retry);
1610            self.spill_file_upkeep();
1611            let popped = self.spill_queue().pop_front();
1612            if let Some(meta) = popped {
1613                self.spill_process(&meta, SpillKind::Evict);
1614                self.spill.in_flight.fetch_sub(1, Ordering::Relaxed);
1615                continue;
1616            }
1617            if self.spill.eager.load(Ordering::Relaxed) && self.back_one() {
1618                continue;
1619            }
1620            // Nothing to evict or back: park. Re-checking emptiness under
1621            // the queue lock closes the lost-wakeup window (hand-offs push
1622            // under this lock before notifying); the timeout backstops
1623            // everything else (fresh inserts, tier growth, lost notifies).
1624            let queue = self.spill_queue();
1625            if queue.is_empty() {
1626                let _ = self
1627                    .spill
1628                    .cv
1629                    .wait_timeout(queue, SPILL_PARK_TIMEOUT)
1630                    .expect("spill queue poisoned");
1631            }
1632        }
1633    }
1634
1635    /// The spill loop's compressed-tier trim. Tier-2 pageouts ride the spill
1636    /// threads: every pass through the loop (job completion, condvar wakeup,
1637    /// park timeout) trims the compressed tier if needed, at a single atomic
1638    /// load when under cap. While the file store's full hint is set, a pass
1639    /// runs only once `hinted_retry` has passed, which then moves one
1640    /// [`SPILL_PARK_TIMEOUT`] ahead.
1641    fn spill_trim(&self, hinted_retry: &mut std::time::Instant) {
1642        if self.full_hint.load(Ordering::Relaxed) {
1643            // Every worker-side `notify_one` wakes a spill thread, and a busy
1644            // spill queue keeps it from parking, so keying the retry on the
1645            // park's timeout would let wakeups drive a pass per job while the
1646            // store can take nothing.
1647            let now = std::time::Instant::now();
1648            if now < *hinted_retry {
1649                return;
1650            }
1651            *hinted_retry = now + SPILL_PARK_TIMEOUT;
1652        }
1653        self.enforce_compressed_cap(Pass::Background);
1654    }
1655
1656    /// Re-derives the file store's capacity and releases warm file store
1657    /// slots, each when due. The first spill thread to see one due claims
1658    /// its interval.
1659    fn spill_file_upkeep(&self) {
1660        let ExtentStore::File(store) = &self.store else {
1661            return;
1662        };
1663        let now = std::time::Instant::now();
1664        let (refresh, release) = {
1665            let mut release = self.file_release();
1666            let refresh = now >= release.refresh_due;
1667            if refresh {
1668                release.refresh_due = now + FILE_CAPACITY_REFRESH_INTERVAL;
1669            }
1670            // A zero interval would release on every spill-loop wakeup and
1671            // punch the warm slots demotions are about to reuse.
1672            let due = release.bytes_per_class != 0
1673                && !release.interval.is_zero()
1674                && release.due.is_some_and(|due| now >= due);
1675            if due {
1676                release.due = now.checked_add(release.interval);
1677            }
1678            (refresh, due.then_some(release.bytes_per_class))
1679        };
1680        if refresh {
1681            store.refresh_capacity();
1682        }
1683        if let Some(bytes_per_class) = release {
1684            store.release_warm(bytes_per_class);
1685        }
1686    }
1687
1688    fn file_release(&self) -> MutexGuard<'_, FileRelease> {
1689        self.file_release.lock().expect("file release poisoned")
1690    }
1691
1692    /// Eagerly compresses one unbacked chunk from the eviction queues into
1693    /// `BackedResident`, returning whether a chunk was backed — `false`
1694    /// means nothing was actionable (queues empty, or the bounded scans
1695    /// found only already-backed, in-flight, contended, or stale entries)
1696    /// and the caller should park rather than rescan. Bands are visited
1697    /// deepest first, mirroring eviction order so the chunks evicted first
1698    /// are the ones whose backing is already pre-paid.
1699    fn back_one(&self) -> bool {
1700        for band in (0..DEPTH_BANDS).rev() {
1701            if self.back_one_from(band) {
1702                return true;
1703            }
1704        }
1705        false
1706    }
1707
1708    /// One bounded backing scan over a single band's queue. Non-actionable
1709    /// entries are requeued or dropped per the same rules budget
1710    /// enforcement uses, except that the second-chance `touched` bit is
1711    /// left alone — backing is not an eviction and must not consume a
1712    /// chunk's reprieve.
1713    fn back_one_from(&self, band: usize) -> bool {
1714        for _ in 0..16 {
1715            let popped = self.queue(band).pop_front();
1716            let Some(weak) = popped else {
1717                return false;
1718            };
1719            let Some(meta) = weak.upgrade() else {
1720                continue;
1721            };
1722            {
1723                let Ok(mut state) = meta.state.try_lock() else {
1724                    self.queue(band).push_back(weak);
1725                    continue;
1726                };
1727                if state.freed {
1728                    continue;
1729                }
1730                match state.residency {
1731                    Residency::Evicted | Residency::Oversize => {
1732                        continue;
1733                    }
1734                    Residency::UnbackedResident => {
1735                        state.residency = Residency::WriteInFlight;
1736                    }
1737                    Residency::BackedResident | Residency::WriteInFlight => {
1738                        self.queue(band).push_back(weak);
1739                        continue;
1740                    }
1741                }
1742            }
1743            self.spill.in_flight.fetch_add(1, Ordering::Relaxed);
1744            self.spill_process(&meta, SpillKind::Back);
1745            self.spill.in_flight.fetch_sub(1, Ordering::Relaxed);
1746            // The chunk remains an eviction candidate (now a cheap one).
1747            self.queue(band).push_back(weak);
1748            return true;
1749        }
1750        false
1751    }
1752
1753    /// Performs (or cancels) one scheduled compression. Runs on spill
1754    /// threads only. Lock discipline: the chunk lock is held only to validate
1755    /// and to commit, never across the compression or the `pageout`
1756    /// reclaim, which are the multi-millisecond costs this path exists to
1757    /// keep off budget-enforcing threads.
1758    fn spill_process(&self, meta: &Arc<ChunkMeta>, kind: SpillKind) {
1759        // Validate under the lock, then release it for the I/O. The slot is
1760        // captured under the lock and remains owned by this chunk for the
1761        // unlocked compression: in `WriteInFlight`, eviction skips the chunk
1762        // and `ChunkHandle::drop` defers slot release to this thread.
1763        let source: Result<u32, Vec<u64>>;
1764        {
1765            let mut state = meta.state();
1766            if state.freed {
1767                // Freed while queued: the deferred cleanup is ours, and the
1768                // chunk dies without ever compressing — the write-behind
1769                // cancellation window. `ChunkHandle::drop` already counted
1770                // the free and the live-chunks decrement.
1771                self.counters
1772                    .spill_cancelled
1773                    .fetch_add(1, Ordering::Relaxed);
1774                self.counters.writes_elided.fetch_add(1, Ordering::Relaxed);
1775                self.release_resident(meta, &mut state);
1776                return;
1777            }
1778            if state.residency != Residency::WriteInFlight {
1779                return;
1780            }
1781            // A typed body is serialized here, under the lock: the
1782            // serialization is one copy, and compressing the copy below needs
1783            // no lock.
1784            source = match (state.slot, state.typed.as_deref()) {
1785                (Some(slot), _) => Ok(slot),
1786                (None, Some(body)) => Err(serialize_typed(body)),
1787                (None, None) => unreachable!("write-in-flight chunk holds a slot or a body"),
1788            };
1789        }
1790        let data = match &source {
1791            // SAFETY: the chunk is live (the queue holds an `Arc`) and in
1792            // `WriteInFlight`, so the slot is not recycled (`ChunkHandle::drop`
1793            // defers slot release to this thread in that state) and its
1794            // contents are immutable; concurrent copy-out reads take the
1795            // state lock and read the slot, but nothing writes it.
1796            Ok(slot) => unsafe { self.slot_data(meta, *slot) },
1797            Err(words) => &words[..],
1798        };
1799        // Spill threads see a steady job stream, so they keep the grown
1800        // compression scratch for the next job.
1801        let extent = Extent::write(&self.extent_arena, data, meta.codec, Scratch::Retain);
1802        let mut state = meta.state();
1803        if state.freed {
1804            // Freed during compression: the extent is garbage; cleanup is
1805            // ours as above. Compression ran, so this is not an elided free.
1806            self.counters
1807                .spill_cancelled
1808                .fetch_add(1, Ordering::Relaxed);
1809            self.release_resident(meta, &mut state);
1810            return;
1811        }
1812        self.commit_extent(meta, &mut state, extent);
1813        match kind {
1814            SpillKind::Back => {
1815                // The slot stays for write-behind: the chunk remains
1816                // readable, and the extent makes a later budget eviction a
1817                // pure page release.
1818                self.counters.eager_backs.fetch_add(1, Ordering::Relaxed);
1819                state.residency = Residency::BackedResident;
1820            }
1821            SpillKind::Evict => {
1822                // `release_slot`'s precondition holds: the state lock is
1823                // held and `!freed` was observed under it.
1824                self.counters
1825                    .evictions_compress
1826                    .fetch_add(1, Ordering::Relaxed);
1827                self.release_resident(meta, &mut state);
1828                state.residency = Residency::Evicted;
1829            }
1830        };
1831        // The fresh resident extent may put the tier over its cap. The spill
1832        // loop trims once per iteration, so a spill thread alternates one
1833        // eviction with one demotion. A trim here as well would demote twice
1834        // per eviction, with `in_flight` still counting this job, which
1835        // pushes evictions inline onto the threads that trip the budget.
1836    }
1837
1838    /// Releases whichever resident form the chunk holds, its slot or its
1839    /// typed body, under [`PoolInner::release_slot`]'s precondition.
1840    fn release_resident(&self, meta: &ChunkMeta, state: &mut ChunkState) {
1841        if state.slot.is_some() {
1842            self.release_slot(meta, state);
1843        } else if let Some(body) = state.typed.take() {
1844            drop(body);
1845            self.counters
1846                .resident_bytes
1847                .fetch_sub(u64::cast_from(meta.len_bytes()), Ordering::Relaxed);
1848        }
1849    }
1850
1851    /// Releases `state`'s slot — slot returned to the region free list,
1852    /// physical pages discarded unless the slot joins the bounded warm pool —
1853    /// and decrements resident bytes. Releasing pages beyond the warm pool is
1854    /// what keeps RSS aligned with the `resident_bytes` gauge the budget
1855    /// enforcer trusts; the warm pool relaxes that alignment by an explicit,
1856    /// bounded amount (`warm_bytes`, capped at a fraction of the budget) so
1857    /// slot reuse faults no pages and skips the kernel's page zeroing.
1858    ///
1859    /// Precondition: the caller holds the chunk's state lock, and no
1860    /// reference into the slot exists — copy-out reads borrow the slot only
1861    /// under that same lock, and a `WriteInFlight` chunk's unlocked
1862    /// compression read belongs to the spill thread, which is the only
1863    /// caller that releases the slot in that state. This is what makes the
1864    /// `dontneed` below sound, and what makes keeping a warm slot's stale
1865    /// contents safe: the slot's next occupant fully overwrites every byte
1866    /// it reads, satisfying the contents-undefined contract either way.
1867    fn release_slot(&self, meta: &ChunkMeta, state: &mut ChunkState) {
1868        let slot = state.slot.take().expect("slotted chunk");
1869        let region = self.region_of(meta);
1870        let warm = self.try_keep_warm(region.class_size());
1871        if !warm {
1872            // SAFETY: no reference into the slot exists (the function-level
1873            // precondition, established under the held state lock).
1874            unsafe {
1875                region::dontneed(region.slot_ptr(slot), region.class_size());
1876            }
1877        }
1878        region.free(slot, warm);
1879        self.counters
1880            .resident_bytes
1881            .fetch_sub(u64::cast_from(meta.len_bytes()), Ordering::Relaxed);
1882    }
1883
1884    /// The warm pool's byte ceiling: an eighth of the budget, clamped at an
1885    /// absolute maximum. The fraction sizes fault amortization at small
1886    /// budgets; the clamp keeps large budgets from parking gigabytes of idle
1887    /// warm slots no fault rate could justify.
1888    fn warm_cap(&self) -> u64 {
1889        (self.budget_bytes.load(Ordering::Relaxed) / 8).min(1 << 30)
1890    }
1891
1892    /// Cools warm free slots until `warm_bytes` falls to the warm cap. A
1893    /// budget shrink lowers the cap, and warm capacity is checked only when
1894    /// a slot is freed, so without this pass slots parked under the old cap
1895    /// would hold their pages until same-class reuse happened to drain them,
1896    /// exactly when the shrink wanted the memory back.
1897    fn trim_warm_pool(&self) {
1898        let mut over = self
1899            .counters
1900            .warm_bytes
1901            .load(Ordering::Relaxed)
1902            .saturating_sub(self.warm_cap());
1903        for region in &self.regions {
1904            if over == 0 {
1905                return;
1906            }
1907            let cooled = u64::cast_from(region.cool_warm_slots(usize::cast_from(over)));
1908            self.counters
1909                .warm_bytes
1910                .fetch_sub(cooled, Ordering::Relaxed);
1911            over = over.saturating_sub(cooled);
1912        }
1913    }
1914
1915    /// Claims warm-pool capacity for a slot of `class_size` bytes, returning
1916    /// whether the slot may keep its pages. The RSS overshoot the warm pool
1917    /// introduces is bounded by [`PoolInner::warm_cap`] and visible as the
1918    /// `warm_bytes` stat.
1919    fn try_keep_warm(&self, class_size: usize) -> bool {
1920        let cap = self.warm_cap();
1921        let class_bytes = u64::cast_from(class_size);
1922        self.counters
1923            .warm_bytes
1924            .try_update(Ordering::Relaxed, Ordering::Relaxed, |cur| {
1925                (cur + class_bytes <= cap).then_some(cur + class_bytes)
1926            })
1927            .is_ok()
1928    }
1929
1930    /// Allocates a slot in `class` for a payload of `len_bytes` with
1931    /// warm-pool accounting (a warm allocation is counted as a reuse and
1932    /// trimmed to the payload), or `None` when the class has no free slot.
1933    fn try_alloc_slot(&self, class: usize, len_bytes: usize) -> Option<u32> {
1934        let (index, warm) = self.regions[class].alloc()?;
1935        if warm {
1936            let class_bytes = u64::cast_from(self.regions[class].class_size());
1937            self.counters
1938                .warm_bytes
1939                .fetch_sub(class_bytes, Ordering::Relaxed);
1940            self.counters.warm_reuses.fetch_add(1, Ordering::Relaxed);
1941            // A warm slot keeps the prior occupant's resident pages, which
1942            // may extend past the new payload while the ledger credits only
1943            // `len_bytes`.
1944            self.trim_slot_tail(class, index, len_bytes);
1945        }
1946        Some(index)
1947    }
1948
1949    /// Releases a slot's pages beyond the first `len_bytes` (rounded up to
1950    /// a page), so a slot reused for a smaller payload does not keep its
1951    /// prior occupant's tail pages resident with no bytes in the ledger to
1952    /// answer for them.
1953    ///
1954    /// Precondition: the caller exclusively owns the slot (freshly
1955    /// allocated, or taken from a victim under the victim's state lock)
1956    /// with no reference into it.
1957    fn trim_slot_tail(&self, class: usize, slot: u32, len_bytes: usize) {
1958        let region = &self.regions[class];
1959        // Hugepage-class slots trim at huge-page granularity: a base-page
1960        // trim would split the slot's `MADV_HUGEPAGE` folios, and khugepaged
1961        // may later re-collapse a partially trimmed range, re-instantiating
1962        // pages the ledger counts as released. Whole-folio trims leave no
1963        // partial folio to split or resurrect.
1964        let granule = if region.class_size() >= region::HUGE_PAGE {
1965            region::HUGE_PAGE
1966        } else {
1967            region::page_size()
1968        };
1969        let keep = len_bytes.next_multiple_of(granule).min(region.class_size());
1970        let tail = region.class_size() - keep;
1971        if tail == 0 {
1972            return;
1973        }
1974        // SAFETY: the caller exclusively owns the slot per the
1975        // precondition, and `keep + tail` is exactly the class size, so the
1976        // range stays within the slot.
1977        unsafe {
1978            region::dontneed(region.slot_ptr(slot).add(keep), tail);
1979        }
1980    }
1981
1982    /// Allocates a slot in `class` for an insert: as
1983    /// [`PoolInner::try_alloc_slot`], with an exhausted class counted as a
1984    /// heap fallback for a `len_bytes` payload (warned about once). `None`
1985    /// means the caller must degrade to the heap.
1986    fn alloc_slot(&self, class: usize, len_bytes: usize) -> Option<u32> {
1987        match self.try_alloc_slot(class, len_bytes) {
1988            Some(index) => Some(index),
1989            None => {
1990                self.counters
1991                    .slot_exhausted_fallbacks
1992                    .fetch_add(1, Ordering::Relaxed);
1993                static EXHAUSTED_ONCE: std::sync::Once = std::sync::Once::new();
1994                EXHAUSTED_ONCE.call_once(|| {
1995                    tracing::warn!(
1996                        len_bytes,
1997                        "buffer pool size class exhausted; falling back to heap chunks \
1998                         (raise the pool's per-class virtual reservation)",
1999                    );
2000                });
2001                None
2002            }
2003        }
2004    }
2005
2006    /// Acquires a slot for re-admitting an evicted chunk, from free budget
2007    /// headroom or by stealing a clean backed victim's slot, never by
2008    /// evicting or compressing anything. `None` counts a denied admission.
2009    /// On success the admitted chunk's resident-bytes accounting and the
2010    /// admission counter are settled, and the caller (who holds the chunk's
2011    /// state lock) owns the slot: its contents are unspecified (fresh,
2012    /// warm, or the victim's stale bytes) and must be fully overwritten.
2013    fn admit_slot(&self, meta: &ChunkMeta) -> Option<u32> {
2014        let class = meta.class.expect("evicted chunk has a class");
2015        let len_bytes = u64::cast_from(meta.len_bytes());
2016        // Free budget first: reserve the bytes, then a slot. The
2017        // reservation never pushes resident bytes past the budget, and a
2018        // class with no free slot hands the reservation back rather than
2019        // evicting anything to make room. The headroom test uses evictable
2020        // bytes, matching budget enforcement: unevictable heap-backed bytes
2021        // must not permanently veto budget-path admissions the enforcer
2022        // would never need to undo.
2023        let reserved = self
2024            .counters
2025            .resident_bytes
2026            .try_update(Ordering::Relaxed, Ordering::Relaxed, |cur| {
2027                // Loaded inside the closure so a CAS retry sees oversize
2028                // frees that landed since the last attempt.
2029                let oversize = self.counters.oversize_bytes.load(Ordering::Relaxed);
2030                let next = cur.checked_add(len_bytes)?;
2031                (next.saturating_sub(oversize) <= self.budget_bytes.load(Ordering::Relaxed))
2032                    .then_some(next)
2033            })
2034            .is_ok();
2035        if reserved {
2036            if let Some(slot) = self.try_alloc_slot(class, meta.len_bytes()) {
2037                self.counters
2038                    .admissions_budget
2039                    .fetch_add(1, Ordering::Relaxed);
2040                return Some(slot);
2041            }
2042            self.counters
2043                .resident_bytes
2044                .fetch_sub(len_bytes, Ordering::Relaxed);
2045        }
2046        if let Some(slot) = self.steal_clean_victim(class, meta.len_bytes()) {
2047            // The slot's physical pages transfer deliberately, but only up
2048            // to the admitted payload: the victim's pages past it would
2049            // stay resident with no ledger bytes to answer for them.
2050            self.trim_slot_tail(class, slot, meta.len_bytes());
2051            self.counters
2052                .admissions_steal
2053                .fetch_add(1, Ordering::Relaxed);
2054            return Some(slot);
2055        }
2056        self.counters
2057            .admissions_denied
2058            .fetch_add(1, Ordering::Relaxed);
2059        None
2060    }
2061
2062    /// Takes the slot of a clean victim in `class` for an admitted payload
2063    /// of `admitted_len_bytes`: a `BackedResident` chunk with a clear
2064    /// touched bit, whose extent already duplicates its slot, so the victim
2065    /// transitions to `Evicted` with zero I/O, its extent intact, and its
2066    /// queue entry dropped. The returned slot keeps its physical pages (no
2067    /// `dontneed`, no free-list round trip); they hold the victim's stale
2068    /// bytes. `None` when the bounded scan finds no such victim, or none
2069    /// whose growth the budget can absorb.
2070    ///
2071    /// The caller holds its own chunk's state lock. The scan follows the
2072    /// enforcement discipline (deepest band first, queue guard dropped
2073    /// before any chunk lock, victims only ever `try_lock`ed), which is
2074    /// what keeps the chunk-lock-while-probing-chunk-lock window
2075    /// deadlock-free: two admitters stealing toward each other both fail
2076    /// the `try_lock` and skip. Unlike enforcement, the scan rotates
2077    /// unsuitable entries (touched, wrong class, unbacked) to the back
2078    /// without spending touched bits, shuffling FIFO order the way the
2079    /// backing scan does.
2080    fn steal_clean_victim(&self, class: usize, admitted_len_bytes: usize) -> Option<u32> {
2081        // Bound on entries examined, per band rather than shared across the
2082        // scan: a deep band dense with touched resident chunks would
2083        // otherwise spend the whole scan on hopeless candidates and starve
2084        // the shallower bands where eager backing stocks the clean victims.
2085        // Eight visits absorb a few lock-busy or freshly touched entries
2086        // without degrading a hopeless scan into a full queue walk.
2087        const VISITS_PER_BAND: usize = 8;
2088        for band in (0..DEPTH_BANDS).rev() {
2089            let mut visits = VISITS_PER_BAND;
2090            while visits > 0 {
2091                let popped = self.queue(band).pop_front();
2092                let Some(weak) = popped else {
2093                    // Band exhausted; the next band has its own budget.
2094                    break;
2095                };
2096                let Some(meta) = weak.upgrade() else {
2097                    // Stale entries drop for free and do not spend a visit.
2098                    continue;
2099                };
2100                visits -= 1;
2101                let Ok(mut state) = meta.state.try_lock() else {
2102                    self.queue(band).push_back(weak);
2103                    continue;
2104                };
2105                if state.freed {
2106                    continue;
2107                }
2108                match state.residency {
2109                    // Entries for non-resident chunks drop, as in
2110                    // enforcement.
2111                    Residency::Evicted | Residency::Oversize => continue,
2112                    Residency::UnbackedResident | Residency::WriteInFlight => {
2113                        self.queue(band).push_back(weak);
2114                        continue;
2115                    }
2116                    Residency::BackedResident => {}
2117                }
2118                // A typed victim has no slot to hand over.
2119                if state.touched || meta.class != Some(class) || state.slot.is_none() {
2120                    self.queue(band).push_back(weak);
2121                    continue;
2122                }
2123                // Settle the ledger in one step: the victim's bytes out, the
2124                // admitted payload's in. A steal that grows resident bytes
2125                // is an admission and must fit the budget (against evictable
2126                // bytes, as everywhere); a shrinking steal always may
2127                // proceed. On failure the victim is requeued untouched.
2128                let victim_len = u64::cast_from(meta.len_bytes());
2129                let admitted_len = u64::cast_from(admitted_len_bytes);
2130                let settled = self
2131                    .counters
2132                    .resident_bytes
2133                    .try_update(Ordering::Relaxed, Ordering::Relaxed, |cur| {
2134                        let next = cur.checked_add(admitted_len)?.saturating_sub(victim_len);
2135                        let oversize = self.counters.oversize_bytes.load(Ordering::Relaxed);
2136                        (next <= cur
2137                            || next.saturating_sub(oversize)
2138                                <= self.budget_bytes.load(Ordering::Relaxed))
2139                        .then_some(next)
2140                    })
2141                    .is_ok();
2142                if !settled {
2143                    self.queue(band).push_back(weak);
2144                    continue;
2145                }
2146                let slot = state.slot.take().expect("backed chunk has a slot");
2147                state.residency = Residency::Evicted;
2148                return Some(slot);
2149            }
2150        }
2151        None
2152    }
2153
2154    /// Capacity of the compressed-resident tier: the RSS target's headroom
2155    /// above the slot budget, the insertion slack currently in use, and the
2156    /// warm cap. With no target set the tier has zero capacity, so extents
2157    /// page out as soon as they are written.
2158    fn compressed_cap(&self) -> u64 {
2159        let target = self.rss_target_bytes.load(Ordering::Relaxed);
2160        let budget = self.budget_bytes.load(Ordering::Relaxed);
2161        let floor = budget
2162            .saturating_add(self.insert_slack_in_use(budget))
2163            .saturating_add(self.warm_cap());
2164        target.saturating_sub(floor)
2165    }
2166
2167    /// Whether the compressed tier holds more than `over` bytes, counted as
2168    /// enforcement counts it for the store.
2169    fn compressed_tier_above(&self, over: u64) -> bool {
2170        match &self.store {
2171            ExtentStore::Swap => self.counters.extent_resident_bytes.load(Ordering::Relaxed) > over,
2172            // Heap-fallback extents are never demoted, so they do not keep a
2173            // pass walking the queue.
2174            ExtentStore::File(_) => self.reclaimable_resident() > over,
2175        }
2176    }
2177
2178    /// Slot bytes reserved above `budget` by insertions that outran
2179    /// enforcement, at most [`insert_slack`]. Oversize payloads live on the
2180    /// heap and are excluded, as in [`PoolInner::reserve_insert`].
2181    fn insert_slack_in_use(&self, budget: u64) -> u64 {
2182        let resident = self.counters.resident_bytes.load(Ordering::Relaxed);
2183        let oversize = self.counters.oversize_bytes.load(Ordering::Relaxed);
2184        resident
2185            .saturating_sub(oversize)
2186            .saturating_sub(budget)
2187            .min(insert_slack(budget))
2188    }
2189
2190    /// Counts a newly resident extent (written, or revived by a read)
2191    /// against the compressed tier. A reclaimable extent additionally
2192    /// enqueues its chunk for RSS-target enforcement; an unreclaimable one
2193    /// (a heap-fallback extent, which is never advised out) counts against
2194    /// the unreclaimable gauge instead and stays out of the queue, so
2195    /// enforcement never walks entries it cannot act on. Callers hold the
2196    /// chunk's state lock with the extent present and resident, and follow
2197    /// up with [`PoolInner::enforce_compressed_cap`] once the lock is
2198    /// released.
2199    ///
2200    /// Invariant: `extent_resident_bytes` equals the sum of `alloc_size`
2201    /// over live chunks' extents whose `is_resident()` is true, and
2202    /// `extent_residents` counts those extents; `extent_unreclaimable_bytes`
2203    /// is the subset whose `pageout_capped()` is true. A freed chunk's
2204    /// `Demoting` extent stays counted until its demoter's commit releases
2205    /// it. This method, [`PoolInner::note_extent_reclaimable`],
2206    /// [`PoolInner::note_extent_released`], the pageout arms in
2207    /// [`PoolInner::enforce_compressed_cap`], and the demotion commit in
2208    /// [`PoolInner::demote`] are the only adjusters. Every flag flip pairs
2209    /// with one of them under the chunk's state lock.
2210    ///
2211    /// Likewise `extent_file_bytes` equals the sum of `file_bytes()` over
2212    /// live chunks' extents. The demotion commit adds to it and
2213    /// [`PoolInner::release_extent`] subtracts, each under the chunk's state
2214    /// lock.
2215    fn note_extent_resident(&self, meta: &Arc<ChunkMeta>, extent_alloc: usize, reclaimable: bool) {
2216        self.counters
2217            .extent_resident_bytes
2218            .fetch_add(u64::cast_from(extent_alloc), Ordering::Relaxed);
2219        self.extent_residents.fetch_add(1, Ordering::Relaxed);
2220        if reclaimable {
2221            self.enqueue_extent(meta, extent_alloc);
2222        } else {
2223            self.counters
2224                .extent_unreclaimable_bytes
2225                .fetch_add(u64::cast_from(extent_alloc), Ordering::Relaxed);
2226        }
2227    }
2228
2229    /// Returns a retry-capped resident extent to the reclaimable set after a
2230    /// read restored its pageout budget: uncounts it from the unreclaimable
2231    /// gauge and re-enqueues its chunk for RSS-target enforcement. The
2232    /// caller holds the chunk's state lock with the extent present, resident,
2233    /// and no longer `pageout_capped()`.
2234    fn note_extent_reclaimable(&self, meta: &Arc<ChunkMeta>, extent_alloc: usize) {
2235        self.counters
2236            .extent_unreclaimable_bytes
2237            .fetch_sub(u64::cast_from(extent_alloc), Ordering::Relaxed);
2238        self.enqueue_extent(meta, extent_alloc);
2239    }
2240
2241    /// Enqueues a reclaimable resident extent for RSS-target enforcement.
2242    fn enqueue_extent(&self, meta: &Arc<ChunkMeta>, extent_alloc: usize) {
2243        self.prune_extent_queue();
2244        self.extent_queue()
2245            .push_back(ExtentEntry::new(meta, extent_alloc));
2246        // Without spill threads only an inline pass would demote this
2247        // extent, and the set hint suppresses those. The probe runs only
2248        // while the hint is set.
2249        if let ExtentStore::File(store) = &self.store {
2250            if self.full_hint.load(Ordering::Relaxed)
2251                && store
2252                    .class_for(extent_alloc)
2253                    .is_some_and(|class| store.can_alloc(class))
2254            {
2255                self.full_hint.store(false, Ordering::Relaxed);
2256            }
2257        }
2258    }
2259
2260    /// Uncounts a resident extent that is being dropped (chunk freed or
2261    /// degraded). Its queue entry goes stale and is dropped on visit or by
2262    /// [`PoolInner::prune_extent_queue`].
2263    fn note_extent_released(&self, extent: &Extent) {
2264        if extent.is_resident() {
2265            self.counters
2266                .extent_resident_bytes
2267                .fetch_sub(u64::cast_from(extent.alloc_size()), Ordering::Relaxed);
2268            self.extent_residents.fetch_sub(1, Ordering::Relaxed);
2269            if extent.pageout_capped() {
2270                self.counters
2271                    .extent_unreclaimable_bytes
2272                    .fetch_sub(u64::cast_from(extent.alloc_size()), Ordering::Relaxed);
2273            }
2274        }
2275    }
2276
2277    /// Drops a freed chunk's extent with its accounting, under the chunk's
2278    /// state lock. A `Demoting` extent stays in place, per `Home::Demoting`.
2279    fn release_extent(&self, state: &mut ChunkState) {
2280        let Some(extent) = &state.extent else {
2281            return;
2282        };
2283        if extent.is_demoting() {
2284            return;
2285        }
2286        if extent.is_reclaimable_arena() {
2287            self.counters
2288                .extent_demotions_elided
2289                .fetch_add(1, Ordering::Relaxed);
2290        }
2291        let file_bytes = extent.file_bytes();
2292        if let Some(file_bytes) = file_bytes {
2293            self.counters
2294                .extent_file_bytes
2295                .fetch_sub(u64::cast_from(file_bytes), Ordering::Relaxed);
2296        }
2297        self.note_extent_released(extent);
2298        // Dropping a `File` extent returns its file slot.
2299        state.extent = None;
2300        if file_bytes.is_some() {
2301            self.full_hint.store(false, Ordering::Relaxed);
2302        }
2303    }
2304
2305    /// Drops extent-queue entries whose chunk has been freed, mirroring
2306    /// [`PoolInner::prune_queues`]: compact only when the queue outgrows
2307    /// all live resident extents (plus a small floor), so the cost
2308    /// amortizes to a constant per push. Enforcement drops stale entries
2309    /// too, but only while the tier is over capacity. A pool that stays
2310    /// under its compressed cap would otherwise accumulate an entry (and a
2311    /// pin on the dead chunk's allocation) per freed extent forever.
2312    fn prune_extent_queue(&self) {
2313        let live = usize::cast_from(self.extent_residents.load(Ordering::Relaxed));
2314        // Parked entries count as queued. Their count is read before the
2315        // queue lock, so the two locks are never held together.
2316        let parked = self.parked_classes.load(Ordering::Relaxed) != 0;
2317        let parked_len = if parked { self.parked_extents().len } else { 0 };
2318        let mut queue = self.extent_queue();
2319        if queue.len() + parked_len <= 2 * live + 16 {
2320            return;
2321        }
2322        queue.retain(|entry| entry.chunk.strong_count() > 0);
2323        drop(queue);
2324        if parked {
2325            let mut parked = self.parked_extents();
2326            let mut len = 0;
2327            for (class, entries) in parked.classes.iter_mut().enumerate() {
2328                entries.retain(|entry| entry.chunk.strong_count() > 0);
2329                if entries.is_empty() {
2330                    self.parked_classes
2331                        .fetch_and(!class_bit(class), Ordering::Relaxed);
2332                }
2333                len += entries.len();
2334            }
2335            parked.len = len;
2336        }
2337    }
2338
2339    /// Sets aside `entry`, whose extent class `class` the store refused,
2340    /// until a pass finds the class placeable. A class beyond the bit mask
2341    /// is requeued at the back instead.
2342    fn park_extent(&self, class: usize, entry: ExtentEntry) {
2343        let bit = class_bit(class);
2344        if bit == 0 {
2345            self.extent_queue().push_back(entry);
2346            return;
2347        }
2348        let mut parked = self.parked_extents();
2349        if parked.classes.len() <= class {
2350            parked.classes.resize_with(class + 1, VecDeque::new);
2351        }
2352        parked.classes[class].push_back(entry);
2353        parked.len += 1;
2354        self.parked_classes.fetch_or(bit, Ordering::Relaxed);
2355    }
2356
2357    /// Returns the parked entries of every class `store` can place now to
2358    /// the front of the extent queue, where they keep their age order ahead
2359    /// of the younger entries queued since.
2360    fn unpark_placeable(&self, store: &FileStore) {
2361        if self.parked_classes.load(Ordering::Relaxed) == 0 {
2362            return;
2363        }
2364        let mut returned = Vec::new();
2365        {
2366            let mut parked = self.parked_extents();
2367            for (class, entries) in parked.classes.iter_mut().enumerate() {
2368                if !entries.is_empty() && store.can_alloc(class) {
2369                    returned.extend(entries.drain(..));
2370                    self.parked_classes
2371                        .fetch_and(!class_bit(class), Ordering::Relaxed);
2372                }
2373            }
2374            parked.len -= returned.len();
2375        }
2376        let mut queue = self.extent_queue();
2377        for entry in returned.into_iter().rev() {
2378            queue.push_front(entry);
2379        }
2380    }
2381
2382    /// Routes compressed-cap enforcement off latency-sensitive threads: with
2383    /// spill threads spawned, wakes one to perform the pageouts
2384    /// (`MADV_PAGEOUT` is synchronous reclaim, bounded per extent but not
2385    /// free at chunk rates). Without them, enforces inline unless the file
2386    /// store's full hint is set. The test is for
2387    /// thread existence, not `spill.enabled`: spawned threads trim the tier
2388    /// in their loop even with eviction hand-off disabled.
2389    ///
2390    /// Deferral makes the target eventually-enforced with bounded lag, and
2391    /// the backstop below bounds the lag: a caller finding the reclaimable
2392    /// tier at double its capacity enforces inline, so sustained creation
2393    /// cannot outrun trimming by more than one capacity's worth. In file
2394    /// mode the bound does not hold while the store's full hint is set or
2395    /// its writes are disabled. The backstop then defers to the spill
2396    /// threads, and the tier grows past the bound by whatever the store
2397    /// cannot take.
2398    fn enforce_or_defer_compressed_cap(&self) {
2399        if self.spill.threads.load(Ordering::Relaxed) > 0 {
2400            // The inline backstop keys on the bytes enforcement can actually
2401            // reclaim. Unreclaimable extents (retry-capped, heap-backed)
2402            // would otherwise hold the backstop permanently over threshold
2403            // and put a full enforcement pass on every caller. For the same
2404            // reason, a file store that cannot take a demotion right now
2405            // leaves the backstop to the spill threads.
2406            let stalled = match &self.store {
2407                ExtentStore::Swap => false,
2408                ExtentStore::File(store) => {
2409                    store.writes_disabled() || self.full_hint.load(Ordering::Relaxed)
2410                }
2411            };
2412            if !stalled && self.reclaimable_resident() > backstop_threshold(self.compressed_cap()) {
2413                self.enforce_compressed_cap(Pass::Inline);
2414            } else {
2415                self.spill.cv.notify_one();
2416            }
2417        } else if !self.full_hint.load(Ordering::Relaxed) {
2418            // Inline enforcement is the only demotion without spill threads,
2419            // and a pass over a store that took nothing last time would
2420            // repeat on every insert and eviction.
2421            self.enforce_compressed_cap(Pass::Inline);
2422        }
2423    }
2424
2425    /// Bytes of resident extents RSS-target enforcement can push out.
2426    fn reclaimable_resident(&self) -> u64 {
2427        let resident = self.counters.extent_resident_bytes.load(Ordering::Relaxed);
2428        let unreclaimable = self
2429            .counters
2430            .extent_unreclaimable_bytes
2431            .load(Ordering::Relaxed);
2432        resident.saturating_sub(unreclaimable)
2433    }
2434
2435    /// Pages out the oldest resident extents until the compressed tier falls
2436    /// to its capacity. The compression is already paid and the device write
2437    /// is the kernel's async writeback, so each pageout is one bounded
2438    /// madvise plus a page-table observation; spill threads run this between
2439    /// jobs, and other threads only when no spill threads exist (see
2440    /// [`PoolInner::enforce_or_defer_compressed_cap`]). Not single-flighted:
2441    /// concurrent passes pop disjoint victims. Visits are bounded by the
2442    /// queue's length at entry; stale entries (extent paged out, dropped, or
2443    /// chunk dead) are dropped. Incomplete extents are requeued with their
2444    /// accounting intact until their retry budget runs out, at which point
2445    /// they leave the queue with their bytes on the unreclaimable gauge, so
2446    /// the tier may settle above its capacity by the bytes the kernel
2447    /// declined to reclaim without enforcement re-walking them.
2448    ///
2449    /// In file mode each victim is demoted to the file store instead, per
2450    /// [`PoolInner::demote`]. A victim whose class the store cannot place
2451    /// moves to the back of the queue, and a store that can place no class
2452    /// or has writes disabled ends the pass. An inline file-mode pass with
2453    /// spill threads present stops at the [`backstop_threshold`], and a
2454    /// background file-mode pass stops per [`DEMOTION_PASS_BYTES`]. An inline
2455    /// file-mode pass with spill threads present also stops after
2456    /// [`DEMOTION_PASS_BYTES`].
2457    fn enforce_compressed_cap(&self, pass: Pass) {
2458        let cap = self.compressed_cap();
2459        let over = match (&self.store, pass) {
2460            // An inline pass leaves the tier at the backstop threshold, and
2461            // the spill threads demote the rest. This bounds a worker's write
2462            // work to the overage beyond the backstop. Without spill threads
2463            // nobody else would demote, so the pass goes down to the cap.
2464            (ExtentStore::File(_), Pass::Inline)
2465                if self.spill.threads.load(Ordering::Relaxed) > 0 =>
2466            {
2467                backstop_threshold(cap)
2468            }
2469            _ => cap,
2470        };
2471        let above = |inner: &PoolInner| inner.compressed_tier_above(over);
2472        // Under-cap is the common case: answer it with one atomic load and
2473        // no queue lock, so frequent callers (the spill loop) stay cheap.
2474        if !above(self) {
2475            return;
2476        }
2477        if let ExtentStore::File(store) = &self.store {
2478            self.unpark_placeable(store);
2479        }
2480        // File mode: whether this pass demoted an extent, whether the store
2481        // refused one, and whether a refusal ended the pass.
2482        let (mut demoted, mut refused, mut stopped) = (false, false, false);
2483        // File mode: the classes the store refused during this pass. Their
2484        // entries are parked without a lock or a store probe, so a pass
2485        // probes the store at most once per class.
2486        let mut refused_classes = 0u64;
2487        let mut remaining = self.extent_queue().len();
2488        let demotion_limit = match pass {
2489            Pass::Background if !self.spill_queue().is_empty() => 1,
2490            Pass::Background => DEMOTION_PASS_BYTES,
2491            Pass::Inline if self.spill.threads.load(Ordering::Relaxed) > 0 => DEMOTION_PASS_BYTES,
2492            // Without spill threads nobody else would demote.
2493            Pass::Inline => u64::MAX,
2494        };
2495        let mut demoted_bytes = 0u64;
2496        while remaining > 0 && above(self) {
2497            if demoted_bytes >= demotion_limit {
2498                break;
2499            }
2500            remaining -= 1;
2501            let popped = self.extent_queue().pop_front();
2502            let Some(entry) = popped else {
2503                break;
2504            };
2505            #[cfg(test)]
2506            testing::EXTENT_VISITS.with(|visits| visits.set(visits.get() + 1));
2507            let Some(meta) = entry.chunk.upgrade() else {
2508                continue;
2509            };
2510            if let ExtentStore::File(store) = &self.store {
2511                if let Some(class) = store.class_for(entry.alloc_size) {
2512                    if refused_classes & class_bit(class) != 0 {
2513                        self.park_extent(class, entry);
2514                        continue;
2515                    }
2516                }
2517            }
2518            // `try_lock`: a chunk mid-read or mid-compression holds its lock
2519            // for milliseconds; requeue rather than convoy behind it.
2520            let Ok(mut state) = meta.state.try_lock() else {
2521                self.extent_queue().push_back(entry);
2522                continue;
2523            };
2524            if let ExtentStore::File(store) = &self.store {
2525                let alloc_size = u64::cast_from(entry.alloc_size);
2526                match self.demote(store, &meta, state, entry, pass) {
2527                    Demotion::Committed => {
2528                        demoted = true;
2529                        demoted_bytes += alloc_size;
2530                    }
2531                    Demotion::Skipped => {}
2532                    Demotion::Refused(class) => {
2533                        refused = true;
2534                        refused_classes |= class_bit(class);
2535                    }
2536                    Demotion::Stopped => {
2537                        (refused, stopped) = (true, true);
2538                        break;
2539                    }
2540                    Demotion::WriteFailed => {
2541                        (refused, stopped) = (true, true);
2542                        break;
2543                    }
2544                }
2545                continue;
2546            }
2547            match &mut state.extent {
2548                Some(extent) if extent.is_resident() => {
2549                    if extent.pageout_capped() {
2550                        // A leftover entry for an already-capped extent (its
2551                        // capping transition below accounted it and dropped
2552                        // its entry): drop this one too. The read that
2553                        // restores the retry budget re-enqueues the chunk.
2554                    } else if extent.pageout() {
2555                        self.counters
2556                            .extent_resident_bytes
2557                            .fetch_sub(u64::cast_from(extent.alloc_size()), Ordering::Relaxed);
2558                        self.extent_residents.fetch_sub(1, Ordering::Relaxed);
2559                        self.counters
2560                            .extent_pageouts
2561                            .fetch_add(1, Ordering::Relaxed);
2562                    } else {
2563                        // The advice left pages resident. The extent keeps
2564                        // its full accounting (the ledger may over-count
2565                        // RSS, the safe direction).
2566                        self.counters
2567                            .extent_pageout_incomplete
2568                            .fetch_add(1, Ordering::Relaxed);
2569                        if extent.pageout_capped() {
2570                            // The retry budget just ran out: the extent
2571                            // leaves the queue and its bytes move to the
2572                            // unreclaimable gauge, so enforcement and the
2573                            // inline backstop stop chasing memory the kernel
2574                            // will not give back. A read that restores the
2575                            // budget re-counts and re-enqueues it.
2576                            self.counters
2577                                .extent_unreclaimable_bytes
2578                                .fetch_add(u64::cast_from(extent.alloc_size()), Ordering::Relaxed);
2579                        } else {
2580                            // Budget remains: keep the queue slot so later
2581                            // passes retry it up to the cap.
2582                            self.extent_queue().push_back(entry);
2583                        }
2584                    }
2585                }
2586                // Paged out already or dropped: the entry is stale. A later
2587                // resident event re-enqueues.
2588                _ => {}
2589            }
2590        }
2591        // Parked extents the store still cannot place keep the tier above
2592        // its cap like a refusal in this pass.
2593        if !refused && self.parked_classes.load(Ordering::Relaxed) != 0 && above(self) {
2594            refused = true;
2595        }
2596        if refused {
2597            self.counters
2598                .extent_file_full
2599                .fetch_add(1, Ordering::Relaxed);
2600            // A pass that demoted something may succeed again on the next
2601            // call, so it leaves the hint alone unless it stopped.
2602            if stopped || !demoted {
2603                self.full_hint.store(true, Ordering::Relaxed);
2604            }
2605        }
2606    }
2607
2608    /// Demotes the extent of the chunk behind `state` to `store`, the file
2609    /// mode counterpart of the pageout arms of
2610    /// [`PoolInner::enforce_compressed_cap`]. `entry` is the chunk's popped
2611    /// queue entry, which this function requeues unless the entry is stale
2612    /// or the extent left the arena.
2613    ///
2614    /// The write runs without the chunk lock, with the extent `Demoting`
2615    /// (see `Home::Demoting` in `extent.rs`).
2616    fn demote(
2617        &self,
2618        store: &Arc<FileStore>,
2619        meta: &Arc<ChunkMeta>,
2620        mut state: MutexGuard<'_, ChunkState>,
2621        entry: ExtentEntry,
2622        pass: Pass,
2623    ) -> Demotion {
2624        let Some(extent) = state.extent.as_mut() else {
2625            // Dropped: the entry is stale.
2626            return Demotion::Skipped;
2627        };
2628        if !extent.is_reclaimable_arena() {
2629            // Demoted, being demoted by another pass, or heap-backed: the
2630            // entry is stale. A failed demotion requeues its own entry.
2631            return Demotion::Skipped;
2632        }
2633        let class = store
2634            .class_for(extent.comp_len())
2635            .expect("the file ladder fits every stored payload");
2636        assert_eq!(
2637            store.class_size(class),
2638            extent.alloc_size(),
2639            "the file store and the arena resolve the extent to different classes",
2640        );
2641        #[cfg(test)]
2642        testing::DEMOTION_PROBES.with(|probes| probes.set(probes.get() + 1));
2643        // NOTE: the allocation runs under the chunk lock and can punch holes
2644        // or `fallocate`, so this lock can be held across those syscalls. The
2645        // lock order permits it, since the store's locks are leaves.
2646        let file_slot = match store.can_alloc(class).then(|| store.alloc(class)) {
2647            Some(Ok(file_slot)) => file_slot,
2648            Some(Err(_)) | None => {
2649                drop(state);
2650                return self.refuse(store, class, entry);
2651            }
2652        };
2653        let source = extent.begin_demotion(file_slot);
2654        let alloc_size = extent.alloc_size();
2655        drop(state);
2656
2657        #[cfg(test)]
2658        testing::run_demotion_hook();
2659        // SAFETY: the extent stays `Demoting` until the commit below, per
2660        // `Home::Demoting`: only this function moves it out of that home.
2661        let (crc, written) = unsafe { source.write(store) };
2662        // Counted here rather than at the commit, so that every write
2663        // counter covers the writes that reached the device, including those
2664        // whose chunk dies before the commit.
2665        if let Ok(write_bytes) = written {
2666            if pass == Pass::Inline {
2667                self.counters
2668                    .extent_file_writes_inline
2669                    .fetch_add(1, Ordering::Relaxed);
2670            }
2671            let by_codec = if meta.codec.is_identity() {
2672                &self.counters.extent_file_write_bytes_identity
2673            } else {
2674                &self.counters.extent_file_write_bytes_compressed
2675            };
2676            by_codec.fetch_add(u64::cast_from(write_bytes), Ordering::Relaxed);
2677        }
2678
2679        let mut state = meta.state();
2680        if state.freed {
2681            // The chunk died mid-write. Its handle left the extent to this
2682            // commit, which returns both slots: the file slot to the store,
2683            // the arena slot through the extent's drop once it is back home.
2684            let mut extent = state.extent.take().expect("demoting extent stays in place");
2685            let file_slot = extent.abort_demotion();
2686            // NOTE: a write that failed `Full` already returned the file
2687            // slot, and freeing it again would list it free twice.
2688            if !matches!(written, Err(WriteError::Full)) {
2689                store.free(file_slot);
2690            }
2691            // A store with writes disabled stays unplaceable however many
2692            // slots are free.
2693            if !matches!(written, Err(WriteError::Disabled)) {
2694                self.full_hint.store(false, Ordering::Relaxed);
2695            }
2696            self.note_extent_released(&extent);
2697            drop(extent);
2698            return Demotion::Skipped;
2699        }
2700        let extent = state
2701            .extent
2702            .as_mut()
2703            .expect("demoting extent stays in place");
2704        match written {
2705            Ok(_) => {
2706                extent.commit_demotion(Arc::clone(store), crc);
2707                let file_bytes = u64::cast_from(store.class_size(class));
2708                self.full_hint.store(false, Ordering::Relaxed);
2709                self.counters
2710                    .extent_resident_bytes
2711                    .fetch_sub(u64::cast_from(alloc_size), Ordering::Relaxed);
2712                self.extent_residents.fetch_sub(1, Ordering::Relaxed);
2713                self.counters
2714                    .extent_pageouts
2715                    .fetch_add(1, Ordering::Relaxed);
2716                self.counters
2717                    .extent_file_bytes
2718                    .fetch_add(file_bytes, Ordering::Relaxed);
2719                Demotion::Committed
2720            }
2721            Err(WriteError::Full) => {
2722                // The store took the file slot back and lowered its
2723                // capacity, so this is a refusal like a failed allocation.
2724                let _ = extent.abort_demotion();
2725                drop(state);
2726                self.refuse(store, class, entry)
2727            }
2728            Err(WriteError::Disabled) => {
2729                // The store disabled its writes and counted the error. The
2730                // extent stays in the arena for good, and its entry returns
2731                // to the front as on a stopped pass.
2732                store.free(extent.abort_demotion());
2733                self.extent_queue().push_front(entry);
2734                Demotion::WriteFailed
2735            }
2736        }
2737    }
2738
2739    /// Parks or requeues the entry of an extent whose `class` the store
2740    /// could not take.
2741    fn refuse(&self, store: &FileStore, class: usize, entry: ExtentEntry) -> Demotion {
2742        // `can_alloc` is per class: a full store may still place smaller
2743        // classes in its free capacity, or any class in a warm slot of that
2744        // class. While some class can allocate, the entry is parked so the
2745        // entries behind it get their turn, and later passes skip it until
2746        // its class fits. Requeued at the back instead, it would cost every
2747        // later pass a visit, which makes each pass linear in the number of
2748        // unplaceable extents. Otherwise no entry can succeed and the pass
2749        // ends, with the entry back at the front: every pass then stops at
2750        // this entry while the store stays full.
2751        if store.can_alloc_any() {
2752            self.park_extent(class, entry);
2753            return Demotion::Refused(class);
2754        }
2755        self.extent_queue().push_front(entry);
2756        Demotion::Stopped
2757    }
2758
2759    /// If the chunk is a live `UnbackedResident` holding a slot and the
2760    /// spill threads have capacity, transitions it to `WriteInFlight` and
2761    /// hands it to them, returning `true`. The hand-off happens under the
2762    /// held state lock; the spill thread blocks on that lock only after this
2763    /// call returns and the caller releases it.
2764    fn spill_handoff(&self, meta: &Arc<ChunkMeta>, state: &mut ChunkState) -> bool {
2765        // The slot check excludes empty chunks, which are `UnbackedResident`
2766        // without a slot: handing one off would panic the spill thread on
2767        // the missing slot.
2768        if state.residency != Residency::UnbackedResident
2769            || state.freed
2770            || (state.slot.is_none() && state.typed.is_none())
2771            || !self.spill_eligible()
2772        {
2773            return false;
2774        }
2775        state.residency = Residency::WriteInFlight;
2776        self.spill_schedule(Arc::clone(meta));
2777        true
2778    }
2779}
2780
2781impl ChunkHandle {
2782    /// Copies the whole contents into `dst` (cleared first), leaving the
2783    /// chunk's residency untouched: a resident slot is copied out directly,
2784    /// and an evicted extent decompresses straight into `dst` without
2785    /// allocating a slot. A read therefore never raises resident bytes,
2786    /// never converts the chunk's state, and hands out no reference into
2787    /// pool memory.
2788    ///
2789    /// The copy runs under the chunk's state lock, which is what makes the
2790    /// no-reference contract cheap: eviction takes the same lock, so there
2791    /// is no reader it could race. The exception is an extent demoted to
2792    /// file, whose read and decode run after the lock is released. The
2793    /// admitting variant is [`ChunkHandle::read_into_admit`].
2794    pub fn read_into(&self, dst: &mut Vec<u64>) {
2795        self.read_impl(0..self.meta.len, dst, false);
2796    }
2797
2798    /// As [`ChunkHandle::read_into`], restricted to the word range `range`
2799    /// of the chunk's contents, which must lie within them. `dst` receives
2800    /// exactly the range.
2801    ///
2802    /// The range narrows only the copy into `dst`: the stored form is a
2803    /// whole compressed block, so a cold read still faults in or reads from
2804    /// file and decompresses the entire extent, and accounting is that of a
2805    /// whole-chunk read.
2806    pub fn read_range_into(&self, range: Range<usize>, dst: &mut Vec<u64>) {
2807        self.read_impl(range, dst, false);
2808    }
2809
2810    /// As [`ChunkHandle::read_into`], except that an evicted chunk is
2811    /// re-admitted to `BackedResident` (its extent kept, its touched bit
2812    /// set) when a slot is available from free budget headroom or by
2813    /// stealing from a clean backed victim of the same size class, never by
2814    /// evicting or compressing anything. When neither source yields a slot
2815    /// the read is served as a plain decompress and the chunk stays
2816    /// evicted.
2817    ///
2818    /// For demand reads on probe paths, where the same chunk is likely to
2819    /// be read again. Merge, drain, and other consume-once paths should use
2820    /// [`ChunkHandle::read_into`] or [`ChunkHandle::take`]: admitting there
2821    /// churns the clean-victim stock that eager backing exists to build,
2822    /// evicting probe targets to house data about to die.
2823    pub fn read_into_admit(&self, dst: &mut Vec<u64>) {
2824        self.read_impl(0..self.meta.len, dst, true);
2825    }
2826
2827    /// As [`ChunkHandle::read_into_admit`], restricted to the word range
2828    /// `range` per [`ChunkHandle::read_range_into`]. Admission is
2829    /// whole-chunk regardless of the range: the acquired slot holds the
2830    /// entire body.
2831    pub fn read_range_into_admit(&self, range: Range<usize>, dst: &mut Vec<u64>) {
2832        self.read_impl(range, dst, true);
2833    }
2834
2835    /// Shared body of the copy-out reads: fills `dst` with the word range
2836    /// `range` of the chunk's contents under the chunk's state lock (the
2837    /// I/O of a plain read of a demoted extent runs after it),
2838    /// re-admitting an evicted chunk when `admit` is set and a slot is
2839    /// available. An empty range returns without locking or touching the
2840    /// chunk, like the whole-chunk read of an empty chunk always has.
2841    fn read_impl(&self, range: Range<usize>, dst: &mut Vec<u64>, admit: bool) {
2842        dst.clear();
2843        let meta = &*self.meta;
2844        assert!(
2845            range.start <= range.end && range.end <= meta.len,
2846            "range {range:?} exceeds the chunk's {} words",
2847            meta.len,
2848        );
2849        if range.is_empty() {
2850            return;
2851        }
2852        let mut state = meta.state();
2853        state.touched = true;
2854        let mut extent_revived = false;
2855        // A plain read of a demoted extent, which runs after the unlock.
2856        let mut unlocked_file_read = None;
2857        match state.residency {
2858            Residency::Oversize => {
2859                let payload = state.oversize.as_ref().expect("oversize chunk has payload");
2860                dst.extend_from_slice(&payload[range.start..range.end]);
2861            }
2862            Residency::Evicted => {
2863                let slot = if admit {
2864                    meta.pool.admit_slot(meta)
2865                } else {
2866                    None
2867                };
2868                let extent = state.extent.as_mut().expect("evicted chunk has an extent");
2869                if extent.file_bytes().is_some() && extent.note_file_read() {
2870                    meta.pool
2871                        .counters
2872                        .extent_file_repeat_reads
2873                        .fetch_add(1, Ordering::Relaxed);
2874                }
2875                if slot.is_none() {
2876                    // Read after the unlock (sound per `PoolInner`'s lock
2877                    // doc), so budget enforcement and admission scans need
2878                    // not skip the chunk while it waits on the device.
2879                    unlocked_file_read = extent.file_location();
2880                }
2881                // Reading faults an in-memory extent's pages back in, so it
2882                // is re-counted against the compressed tier below. A file
2883                // read leaves the extent on file.
2884                let was_resident = extent.is_resident();
2885                let was_capped = extent.pageout_capped();
2886                let extent_alloc = extent.alloc_size();
2887                match slot {
2888                    None if unlocked_file_read.is_some() => {}
2889                    Some(slot) => {
2890                        // Admission: the extent decompresses straight into
2891                        // the acquired slot, fully overwriting its
2892                        // unspecified prior contents, and the caller's
2893                        // buffer is filled from the slot.
2894                        let region = meta.pool.region_of(meta);
2895                        // SAFETY: the slot was acquired for this chunk
2896                        // under its held state lock (freshly allocated, or
2897                        // transferred from the victim under the victim's
2898                        // lock), so it is exclusively owned with no other
2899                        // reference into it, and `len_bytes` fits the
2900                        // class.
2901                        let slot_bytes = unsafe {
2902                            std::slice::from_raw_parts_mut(region.slot_ptr(slot), meta.len_bytes())
2903                        };
2904                        extent.read_into(meta.codec, slot_bytes);
2905                        state.slot = Some(slot);
2906                        state.residency = Residency::BackedResident;
2907                        // SAFETY: the slot belongs to this chunk while the
2908                        // state lock is held (eviction and free both take
2909                        // it).
2910                        let src = unsafe { meta.pool.slot_data(meta, slot) };
2911                        dst.extend_from_slice(&src[range.start..range.end]);
2912                        // Resident again: rejoin the eviction candidates.
2913                        // A leftover entry from before the chunk's eviction
2914                        // stays sound (each entry is validated against the
2915                        // chunk's state on visit) but costs policy: two live
2916                        // entries give the enforcer two chances to spend
2917                        // this chunk's single touched bit, halving its
2918                        // second chance until one entry drains.
2919                        meta.pool
2920                            .queue(band(meta.depth))
2921                            .push_back(Arc::downgrade(&self.meta));
2922                    }
2923                    None => {
2924                        // The zero-fill ahead of the decompress is deliberate
2925                        // waste (~a tenth of the decompress cost): the extent
2926                        // read takes an initialized `&mut [u8]`, so skipping
2927                        // the fill would mean exposing uninitialized memory
2928                        // through a safe reference.
2929                        dst.resize(range.end - range.start, 0);
2930                        let bytes: &mut [u8] = bytemuck::cast_slice_mut(dst.as_mut_slice());
2931                        extent.read_range_into(
2932                            meta.codec,
2933                            meta.len_bytes(),
2934                            range.start * WORD_BYTES,
2935                            bytes,
2936                        );
2937                    }
2938                }
2939                // TODO: a sub-range read of a rangeable stored form (a
2940                // sub-block-framed codec) revives only part of the extent.
2941                // The whole-extent accounting below would then overcount and
2942                // needs a partial-revival variant.
2943                let is_resident = state
2944                    .extent
2945                    .as_ref()
2946                    .expect("evicted chunk has an extent")
2947                    .is_resident();
2948                if !was_resident && is_resident {
2949                    // Revived from the device: the decompress reset any
2950                    // retry budget, so the extent re-enters reclaimable.
2951                    meta.pool
2952                        .note_extent_resident(&self.meta, extent_alloc, true);
2953                    extent_revived = true;
2954                } else if was_capped {
2955                    // The decompress faulted every page and reset the
2956                    // pageout retry budget, so a retry-capped extent is
2957                    // reclaimable again. Heap-backed extents stay
2958                    // structurally capped and stay out of the queue.
2959                    let capped = state
2960                        .extent
2961                        .as_ref()
2962                        .expect("evicted chunk has an extent")
2963                        .pageout_capped();
2964                    if !capped {
2965                        meta.pool.note_extent_reclaimable(&self.meta, extent_alloc);
2966                        extent_revived = true;
2967                    }
2968                }
2969            }
2970            Residency::UnbackedResident | Residency::BackedResident | Residency::WriteInFlight => {
2971                match (state.slot, state.typed.as_deref()) {
2972                    (Some(slot), _) => {
2973                        // SAFETY: the slot belongs to this chunk while the
2974                        // state lock is held (eviction and free both take
2975                        // it).
2976                        let src = unsafe { meta.pool.slot_data(meta, slot) };
2977                        dst.extend_from_slice(&src[range.start..range.end]);
2978                    }
2979                    (None, Some(body)) if range == (0..meta.len) => {
2980                        dst.resize(meta.len, 0);
2981                        body.serialize_into(dst);
2982                    }
2983                    (None, Some(body)) => {
2984                        dst.extend_from_slice(&serialize_typed(body)[range.start..range.end]);
2985                    }
2986                    (None, None) => unreachable!("resident non-empty chunk holds a slot or a body"),
2987                }
2988            }
2989        }
2990        drop(state);
2991        if let Some(location) = unlocked_file_read {
2992            // The zero fill is deliberate, as in the locked plain read.
2993            dst.resize(range.end - range.start, 0);
2994            let bytes: &mut [u8] = bytemuck::cast_slice_mut(dst.as_mut_slice());
2995            location.read_range_into(meta.codec, meta.len_bytes(), range.start * 8, bytes);
2996        }
2997        // The read revived the extent's compressed pages; the tier may need
2998        // trimming. Enforcement locks chunk states itself, so it must run
2999        // after the unlock.
3000        if extent_revived {
3001            meta.pool.enforce_or_defer_compressed_cap();
3002        }
3003    }
3004
3005    /// Runs `f` on the chunk's typed body while it is resident, under the
3006    /// chunk's state lock, or on `None` without the lock in every other
3007    /// state. Eviction only ever `try_lock`s a chunk, so a borrowed body is
3008    /// skipped rather than waited on.
3009    ///
3010    /// NOTE: `f` must not touch this chunk through the pool. The state lock
3011    /// is not reentrant, so a read of the same chunk inside `f` deadlocks.
3012    pub fn with_typed<X>(&self, f: impl FnOnce(Option<&dyn TypedBody>) -> X) -> X {
3013        let mut state = self.meta.state();
3014        if state.typed.is_some() {
3015            state.touched = true;
3016            return f(state.typed.as_deref());
3017        }
3018        drop(state);
3019        f(None)
3020    }
3021
3022    /// Whether the chunk currently holds a resident typed body. Advisory: a
3023    /// concurrent eviction can change the answer as soon as it returns.
3024    pub fn is_typed(&self) -> bool {
3025        self.meta.state().typed.is_some()
3026    }
3027
3028    /// Frees the chunk and returns its typed body without a copy when the
3029    /// body is resident and no spill thread is serializing it, else returns
3030    /// the handle unchanged.
3031    pub fn take_typed(self) -> Result<Box<dyn TypedBody>, ChunkHandle> {
3032        let pool = &self.meta.pool;
3033        let body = {
3034            let mut state = self.meta.state();
3035            let body = match state.residency {
3036                Residency::UnbackedResident => state.typed.take(),
3037                // A demoting extent stays with its demoter until the commit,
3038                // which expects to find it, so the body stays too.
3039                Residency::BackedResident
3040                    if state.typed.is_some()
3041                        && !state.extent.as_ref().is_some_and(Extent::is_demoting) =>
3042                {
3043                    pool.release_extent(&mut state);
3044                    state.residency = Residency::UnbackedResident;
3045                    state.typed.take()
3046                }
3047                Residency::BackedResident
3048                | Residency::WriteInFlight
3049                | Residency::Evicted
3050                | Residency::Oversize => None,
3051            };
3052            if body.is_some() {
3053                pool.counters
3054                    .resident_bytes
3055                    .fetch_sub(u64::cast_from(self.meta.len_bytes()), Ordering::Relaxed);
3056                // Mark the chunk dead before the lock drops. Eager backing
3057                // otherwise sees an `UnbackedResident` chunk with no body,
3058                // moves it to `WriteInFlight`, and the spill thread panics
3059                // finding nothing to serialize.
3060                state.freed = true;
3061            }
3062            body
3063        };
3064        // The emptied chunk drops as an unbacked chunk holding nothing.
3065        body.ok_or(self)
3066    }
3067
3068    /// Copies the whole contents into `dst` (per [`ChunkHandle::read_into`],
3069    /// never admitting) and frees the chunk, cancelling any in-flight
3070    /// backing write.
3071    pub fn take(self, dst: &mut Vec<u64>) {
3072        self.read_into(dst);
3073    }
3074
3075    /// Advisory a consumer may issue before a bulk read: hints the kernel to
3076    /// swap an evicted chunk's extent back in, and is a no-op in every other
3077    /// state and for extents demoted or being demoted to file. Never blocks
3078    /// on I/O (`MADV_WILLNEED` is asynchronous).
3079    pub fn prefetch(&self) {
3080        let state = self.meta.state();
3081        if state.residency == Residency::Evicted {
3082            let extent = state.extent.as_ref().expect("evicted chunk has an extent");
3083            extent.prefetch();
3084        }
3085    }
3086
3087    /// As [`ChunkHandle::prefetch`], scoped to the word range `range` of the
3088    /// chunk's contents. The range is advisory: a backend hints at whatever
3089    /// granularity its stored form permits, and the stored form is a whole
3090    /// compressed block, so it hints the entire extent.
3091    pub fn prefetch_range(&self, range: Range<usize>) {
3092        let _ = range;
3093        self.prefetch();
3094    }
3095}
3096
3097impl Drop for ChunkHandle {
3098    fn drop(&mut self) {
3099        let pool = &self.meta.pool;
3100        let mut state = self.meta.state();
3101        pool.counters.frees.fetch_add(1, Ordering::Relaxed);
3102        state.freed = true;
3103        if self.meta.class.is_some() {
3104            pool.live_chunks.fetch_sub(1, Ordering::Relaxed);
3105        }
3106        let len_bytes = u64::cast_from(self.meta.len_bytes());
3107        // `release_slot`'s precondition holds in every arm below: the handle
3108        // is being dropped, so no copy-out read (which borrows the handle)
3109        // is in progress, and `freed` was set under the state lock held
3110        // here, so concurrent queue visitors skip the chunk.
3111        match state.residency {
3112            Residency::UnbackedResident => {
3113                if state.slot.is_some() || state.typed.is_some() {
3114                    pool.counters.writes_elided.fetch_add(1, Ordering::Relaxed);
3115                    pool.release_resident(&self.meta, &mut state);
3116                }
3117            }
3118            Residency::BackedResident => {
3119                pool.release_resident(&self.meta, &mut state);
3120                pool.release_extent(&mut state);
3121            }
3122            Residency::Evicted => {
3123                crate::soft_assert_no_log!(state.slot.is_none(), "evicted chunk holds no slot");
3124                pool.release_extent(&mut state);
3125            }
3126            Residency::WriteInFlight => {
3127                // A spill thread may be reading the slot to compress it.
3128                // `freed` (set above) tells it the chunk died; it owns the
3129                // slot release, the `resident_bytes` decrement, and the
3130                // cancellation accounting from here.
3131            }
3132            Residency::Oversize => {
3133                pool.counters
3134                    .resident_bytes
3135                    .fetch_sub(len_bytes, Ordering::Relaxed);
3136                pool.counters
3137                    .oversize_bytes
3138                    .fetch_sub(len_bytes, Ordering::Relaxed);
3139                state.oversize = None;
3140            }
3141        }
3142    }
3143}
3144
3145#[cfg(test)]
3146mod tests {
3147    use super::*;
3148    use crate::pool::extent::testing::TEST_CODEC;
3149    use crate::pool::testing::ENFORCE_BUDGET_HOOK;
3150
3151    /// Keep test pools small: 64 MiB of virtual reservation per class.
3152    /// Under Miri the backing is real interpreter heap rather than lazy
3153    /// virtual memory, so shrink further. Classes above the capacity yield
3154    /// empty regions whose inserts degrade to the heap fallback, which is
3155    /// fine: slotted-chunk tests exercise only the smallest classes.
3156    fn test_pool(budget_bytes: usize) -> Pool {
3157        let capacity = if cfg!(miri) { 1 << 20 } else { 64 << 20 };
3158        let pool = Pool::with_class_capacity(capacity).expect("pool creation");
3159        pool.set_budget(budget_bytes);
3160        pool
3161    }
3162
3163    /// Scales an iteration count down under Miri, where one interpreted
3164    /// compression costs what thousands do natively.
3165    fn rounds(native: u64, miri: u64) -> u64 {
3166        if cfg!(miri) { miri } else { native }
3167    }
3168
3169    fn payload(words: usize, seed: u64) -> Vec<u64> {
3170        (0..u64::cast_from(words))
3171            .map(|i| seed.wrapping_mul(0x9E3779B97F4A7C15).wrapping_add(i))
3172            .collect()
3173    }
3174
3175    /// Copies `data` into the pool and clears it.
3176    fn insert(pool: &Pool, data: &mut Vec<u64>) -> ChunkHandle {
3177        insert_at_depth(pool, 0, data)
3178    }
3179
3180    /// Copies `data` into the pool at a hinted depth and clears it.
3181    fn insert_at_depth(pool: &Pool, depth: u8, data: &mut Vec<u64>) -> ChunkHandle {
3182        let hints = ChunkHints { depth };
3183        let handle = pool.insert_with(data.len(), hints, &TEST_CODEC, |dst| {
3184            dst.copy_from_slice(data.as_slice())
3185        });
3186        data.clear();
3187        handle
3188    }
3189
3190    /// Copies a chunk's contents out into a fresh buffer.
3191    fn read(handle: &ChunkHandle) -> Vec<u64> {
3192        let mut out = Vec::new();
3193        handle.read_into(&mut out);
3194        out
3195    }
3196
3197    /// Copies a chunk's contents out into a fresh buffer via the admitting
3198    /// read.
3199    fn read_admit(handle: &ChunkHandle) -> Vec<u64> {
3200        let mut out = Vec::new();
3201        handle.read_into_admit(&mut out);
3202        out
3203    }
3204
3205    /// Words that fill a 64 KiB class exactly.
3206    const SMALL: usize = (64 << 10) / WORD_BYTES;
3207
3208    #[allow(dead_code)]
3209    fn assert_handle_send_sync() {
3210        fn check<T: Send + Sync>() {}
3211        check::<Pool>();
3212        check::<ChunkHandle>();
3213    }
3214
3215    /// The public above-cap predicate follows the tier: false under the cap,
3216    /// true when the kernel declines the pageouts that would restore it.
3217    #[mz_ore::test]
3218    fn compressed_tier_above_cap_tracks_the_tier() {
3219        let pool = test_pool(256 << 20);
3220        pool.set_rss_target(1 << 30);
3221        let handle = insert(&pool, &mut payload(SMALL, 22));
3222        pool.evict(&handle);
3223        assert!(pool.stats().extent_resident_bytes > 0);
3224        assert!(!pool.compressed_tier_above_cap());
3225
3226        region::fake_residency::decline_next(u64::MAX);
3227        pool.set_rss_target(0);
3228        assert_eq!(pool.compressed_cap(), 0);
3229        assert!(
3230            pool.compressed_tier_above_cap(),
3231            "declined pageouts leave the tier above the cap"
3232        );
3233        region::fake_residency::decline_next(0);
3234        drop(handle);
3235        assert!(!pool.compressed_tier_above_cap());
3236    }
3237
3238    /// With an RSS target set, evicted chunks keep their extents resident
3239    /// (the compressed tier); shrinking the target pages the oldest extents
3240    /// out; reads revive them and re-count them.
3241    #[mz_ore::test]
3242    fn compressed_tier_round_trip() {
3243        let pool = test_pool(256 << 20);
3244        pool.set_rss_target(1 << 30);
3245        let orig = payload(SMALL, 21);
3246        let handle = insert(&pool, &mut orig.clone());
3247        pool.evict(&handle);
3248        assert_eq!(handle.residency(), Residency::Evicted);
3249        let stats = pool.stats();
3250        assert!(
3251            stats.extent_resident_bytes > 0,
3252            "under the target, the extent stays resident",
3253        );
3254        assert_eq!(stats.extent_pageouts, 0);
3255
3256        // Shrinking the target to zero pages the extent out.
3257        pool.set_rss_target(0);
3258        let stats = pool.stats();
3259        assert_eq!(stats.extent_resident_bytes, 0, "tier collapsed");
3260        assert_eq!(stats.extent_pageouts, 1);
3261
3262        // Reading revives the extent: contents round-trip, the chunk stays
3263        // evicted, and with the target restored the revived extent is
3264        // counted again.
3265        pool.set_rss_target(1 << 30);
3266        assert_eq!(read(&handle), orig);
3267        assert_eq!(handle.residency(), Residency::Evicted);
3268        assert!(
3269            pool.stats().extent_resident_bytes > 0,
3270            "revived and counted"
3271        );
3272
3273        // Dropping the handle uncounts the resident extent.
3274        drop(handle);
3275        assert_eq!(pool.stats().extent_resident_bytes, 0);
3276    }
3277
3278    /// An RSS target with 1 MiB of headroom above the budget and warm cap
3279    /// keeps an extent resident: unused insertion slack does not shrink the
3280    /// compressed tier.
3281    #[mz_ore::test]
3282    fn unused_insert_slack_leaves_compressed_tier_intact() {
3283        let budget = 64 << 20;
3284        let pool = test_pool(budget);
3285        pool.set_rss_target(budget + budget / 8 + (1 << 20));
3286        let handle = insert(&pool, &mut payload(SMALL, 22));
3287        pool.evict(&handle);
3288        let stats = pool.stats();
3289        assert!(stats.extent_resident_bytes > 0, "the extent stays resident");
3290        assert_eq!(stats.extent_pageouts, 0);
3291    }
3292
3293    /// A ranged read returns exactly the corresponding slice of a
3294    /// whole-chunk read in every residency state, and changes residency
3295    /// exactly as the equivalent whole-chunk read would.
3296    #[mz_ore::test]
3297    fn ranged_reads_match_full_read_slice() {
3298        let pool = test_pool(256 << 20);
3299        pool.set_rss_target(1 << 30);
3300        let orig = payload(SMALL, 33);
3301        let handle = insert(&pool, &mut orig.clone());
3302        let ranges = [
3303            (0usize, 7usize),
3304            (13, 100),
3305            (SMALL - 9, 9),
3306            (0, SMALL),
3307            (5, 0),
3308        ];
3309        let check = |label: &str| {
3310            for (start, len) in ranges {
3311                let mut out = Vec::new();
3312                handle.read_range_into(start..start + len, &mut out);
3313                assert_eq!(
3314                    out,
3315                    &orig[start..start + len],
3316                    "{label} range ({start}, {len})"
3317                );
3318            }
3319        };
3320        assert_eq!(handle.residency(), Residency::UnbackedResident);
3321        check("resident");
3322        pool.evict(&handle);
3323        assert_eq!(handle.residency(), Residency::Evicted);
3324        check("evicted");
3325        assert_eq!(
3326            handle.residency(),
3327            Residency::Evicted,
3328            "plain ranged reads do not admit"
3329        );
3330        // An admitting ranged read returns the range and admits the whole
3331        // chunk.
3332        let mut out = Vec::new();
3333        handle.read_range_into_admit(3..19, &mut out);
3334        assert_eq!(out, &orig[3..19]);
3335        assert_eq!(handle.residency(), Residency::BackedResident);
3336        check("backed");
3337    }
3338
3339    #[mz_ore::test]
3340    #[should_panic(expected = "exceeds the chunk's")]
3341    fn ranged_read_out_of_bounds_panics() {
3342        let pool = test_pool(256 << 20);
3343        let handle = insert(&pool, &mut payload(SMALL, 34));
3344        let mut out = Vec::new();
3345        handle.read_range_into(SMALL - 1..SMALL + 1, &mut out);
3346    }
3347
3348    #[mz_ore::test]
3349    fn default_target_pages_extents_immediately() {
3350        let pool = test_pool(256 << 20);
3351        let handle = insert(&pool, &mut payload(SMALL, 22));
3352        pool.evict(&handle);
3353        let stats = pool.stats();
3354        assert_eq!(stats.extent_resident_bytes, 0);
3355        assert_eq!(stats.extent_pageouts, 1);
3356    }
3357
3358    #[mz_ore::test]
3359    fn full_pageout_uncounts_exactly_the_extent() {
3360        let pool = test_pool(256 << 20);
3361        pool.set_rss_target(1 << 30);
3362        let handle = insert(&pool, &mut payload(SMALL, 50));
3363        pool.evict(&handle);
3364        let counted = pool.stats().extent_resident_bytes;
3365        assert!(counted > 0, "under the target, the extent stays counted");
3366        pool.set_rss_target(0);
3367        let stats = pool.stats();
3368        assert_eq!(stats.extent_resident_bytes, 0, "exactly `counted` left");
3369        assert_eq!(stats.extent_pageouts, 1);
3370        assert_eq!(stats.extent_pageout_incomplete, 0);
3371    }
3372
3373    #[mz_ore::test]
3374    fn incomplete_pageout_keeps_accounting_and_queue_position() {
3375        let pool = test_pool(256 << 20);
3376        pool.set_rss_target(1 << 30);
3377        let handle = insert(&pool, &mut payload(SMALL, 51));
3378        pool.evict(&handle);
3379        let counted = pool.stats().extent_resident_bytes;
3380        assert!(counted > 0);
3381        region::fake_residency::decline_next(1);
3382        pool.set_rss_target(0);
3383        let stats = pool.stats();
3384        assert_eq!(
3385            stats.extent_resident_bytes, counted,
3386            "full accounting stays"
3387        );
3388        assert_eq!(stats.extent_pageouts, 0);
3389        assert_eq!(stats.extent_pageout_incomplete, 1);
3390        assert_eq!(handle.residency(), Residency::Evicted);
3391        // The requeued entry is retried by the next enforcement pass.
3392        pool.enforce_rss_target();
3393        let stats = pool.stats();
3394        assert_eq!(stats.extent_resident_bytes, 0);
3395        assert_eq!(stats.extent_pageouts, 1);
3396        assert_eq!(stats.extent_pageout_incomplete, 1);
3397    }
3398
3399    /// A never-reclaimable extent stops being advised after the retry cap:
3400    /// the incomplete counter stops climbing, the bytes stay counted
3401    /// resident, and the tier keeps paging other extents out around it.
3402    #[mz_ore::test]
3403    fn pageout_retry_cap_stops_advising() {
3404        let pool = test_pool(256 << 20);
3405        let handle = insert(&pool, &mut payload(SMALL, 52));
3406        region::fake_residency::decline_next(u64::MAX);
3407        // RSS target zero: the eviction's enforcement pass advises at once.
3408        pool.evict(&handle);
3409        for _ in 0..5 {
3410            pool.enforce_rss_target();
3411        }
3412        let stats = pool.stats();
3413        assert_eq!(
3414            stats.extent_pageout_incomplete,
3415            u64::from(extent::PAGEOUT_RETRY_CAP),
3416            "advised exactly retry-cap times",
3417        );
3418        assert_eq!(stats.extent_pageouts, 0);
3419        let counted = stats.extent_resident_bytes;
3420        assert!(counted > 0, "capped extent stays counted resident");
3421        // The tier functions around the capped extent: a fresh extent still
3422        // pages out.
3423        region::fake_residency::decline_next(0);
3424        let other = insert(&pool, &mut payload(SMALL, 53));
3425        pool.evict(&other);
3426        let stats = pool.stats();
3427        assert_eq!(stats.extent_pageouts, 1);
3428        assert_eq!(
3429            stats.extent_resident_bytes, counted,
3430            "only the capped extent remains counted",
3431        );
3432        assert_eq!(read(&handle).len(), SMALL, "capped extent stays readable");
3433    }
3434
3435    #[mz_ore::test]
3436    fn read_resets_pageout_retry_budget() {
3437        let pool = test_pool(256 << 20);
3438        let orig = payload(SMALL, 54);
3439        let handle = insert(&pool, &mut orig.clone());
3440        region::fake_residency::decline_next(u64::MAX);
3441        pool.evict(&handle);
3442        for _ in 0..4 {
3443            pool.enforce_rss_target();
3444        }
3445        assert_eq!(
3446            pool.stats().extent_pageout_incomplete,
3447            u64::from(extent::PAGEOUT_RETRY_CAP),
3448            "capped",
3449        );
3450        assert!(pool.stats().extent_resident_bytes > 0);
3451        region::fake_residency::decline_next(0);
3452        assert_eq!(read(&handle), orig);
3453        pool.enforce_rss_target();
3454        let stats = pool.stats();
3455        assert_eq!(stats.extent_pageouts, 1, "the budget reset re-advised it");
3456        assert_eq!(stats.extent_resident_bytes, 0);
3457        // The paged-out extent still round-trips.
3458        assert_eq!(read(&handle), orig);
3459    }
3460
3461    /// Eager backing compresses a chunk to `BackedResident` while it stays
3462    /// readable in its slot; the later budget-driven eviction is a pure page
3463    /// release, and the contents round-trip through the extent.
3464    #[mz_ore::test]
3465    fn eager_backing_round_trip() {
3466        let pool = test_pool(256 << 20);
3467        let orig = payload(SMALL, 11);
3468        let handle = insert(&pool, &mut orig.clone());
3469        assert_eq!(handle.residency(), Residency::UnbackedResident);
3470
3471        assert!(pool.back_step(), "one chunk is backable");
3472        assert_eq!(handle.residency(), Residency::BackedResident);
3473        let stats = pool.stats();
3474        assert_eq!(stats.eager_backs, 1);
3475        assert_eq!(stats.evictions_compress, 0, "backing is not an eviction");
3476        assert!(stats.extent_bytes_written > 0);
3477
3478        // Still readable straight from the slot: the chunk is resident.
3479        assert_eq!(read(&handle), orig);
3480
3481        // The pre-paid eviction is cheap, and the extent round-trips.
3482        pool.evict(&handle);
3483        assert_eq!(handle.residency(), Residency::Evicted);
3484        assert_eq!(pool.stats().evictions_cheap, 1);
3485        pool.poison_free_slots();
3486        assert_eq!(read(&handle), orig);
3487    }
3488
3489    #[mz_ore::test]
3490    fn backing_reports_no_progress_when_all_backed() {
3491        let pool = test_pool(256 << 20);
3492        let _handle = insert(&pool, &mut payload(SMALL, 31));
3493        assert!(pool.back_step(), "one unbacked chunk is actionable");
3494        assert!(!pool.back_step(), "fully backed: no progress");
3495        assert_eq!(pool.stats().eager_backs, 1);
3496    }
3497
3498    /// Freeing under the warm cap parks the slot warm; the next insert of the
3499    /// same class reuses it fault-free and the accounting balances.
3500    #[mz_ore::test]
3501    fn warm_slot_reuse() {
3502        // Budget 8 MiB: warm cap = 1 MiB, so a 64 KiB slot fits warm.
3503        let pool = test_pool(8 << 20);
3504        let orig = payload(SMALL, 7);
3505        let handle = insert(&pool, &mut orig.clone());
3506        drop(handle);
3507        let after_free = pool.stats();
3508        assert_eq!(after_free.warm_bytes, 64 << 10, "freed slot parks warm");
3509        assert_eq!(after_free.warm_reuses, 0);
3510
3511        let handle = insert(&pool, &mut orig.clone());
3512        let after_reuse = pool.stats();
3513        assert_eq!(after_reuse.warm_reuses, 1, "second insert reuses warm slot");
3514        assert_eq!(after_reuse.warm_bytes, 0, "reuse drains the warm pool");
3515        // Contents are correct despite the skipped page release.
3516        assert_eq!(read(&handle), orig);
3517    }
3518
3519    #[mz_ore::test]
3520    fn warm_pool_respects_cap() {
3521        // Budget 1 MiB: warm cap = 128 KiB = two 64 KiB slots.
3522        let pool = test_pool(1 << 20);
3523        let handles: Vec<_> = (0..4)
3524            .map(|seed| insert(&pool, &mut payload(SMALL, seed)))
3525            .collect();
3526        drop(handles);
3527        let stats = pool.stats();
3528        assert_eq!(
3529            stats.warm_bytes,
3530            128 << 10,
3531            "warm pool stops at the budget/8 cap",
3532        );
3533    }
3534
3535    /// A kernel that keeps declining the reclaim advice caps the extent's
3536    /// retry budget: the extent leaves the enforcement queue and moves to
3537    /// the unreclaimable gauge, so enforcement stops walking it, and a read
3538    /// that restores the budget makes it reclaimable and pageable again.
3539    #[mz_ore::test]
3540    fn capped_extents_leave_the_enforcement_queue() {
3541        let pool = test_pool(256 << 20);
3542        let orig = payload(SMALL, 960);
3543        let handle = insert(&pool, &mut orig.clone());
3544        // Decline every observation: eviction's own enforcement pass plus
3545        // the passes below spend the whole retry budget.
3546        region::fake_residency::decline_next(u64::from(extent::PAGEOUT_RETRY_CAP));
3547        pool.evict(&handle);
3548        for _ in 0..extent::PAGEOUT_RETRY_CAP {
3549            pool.enforce_compressed();
3550        }
3551        let stats = pool.stats();
3552        assert_eq!(
3553            stats.extent_pageout_incomplete,
3554            u64::from(extent::PAGEOUT_RETRY_CAP),
3555        );
3556        assert!(stats.extent_unreclaimable_bytes > 0, "capped bytes counted");
3557        assert_eq!(pool.extent_queue_len(), 0, "capped extents leave the queue",);
3558        // Further enforcement is a no-op: nothing queued, no advice spent.
3559        pool.enforce_compressed();
3560        assert_eq!(
3561            pool.stats().extent_pageout_incomplete,
3562            u64::from(extent::PAGEOUT_RETRY_CAP),
3563        );
3564
3565        // A read faults everything back in and restores the retry budget:
3566        // the extent re-enters the reclaimable set and, with the kernel now
3567        // cooperating, the read's own enforcement pass pages it out.
3568        assert_eq!(read(&handle), orig);
3569        let stats = pool.stats();
3570        assert_eq!(stats.extent_unreclaimable_bytes, 0, "budget restored");
3571        assert_eq!(stats.extent_pageouts, 1, "re-enqueued extent pages out");
3572        assert_eq!(stats.extent_resident_bytes, 0);
3573        assert_eq!(pool.extent_queue_len(), 0);
3574    }
3575
3576    /// Shrinking the budget cools warm slots parked under the old, larger
3577    /// cap: their pages are released and `warm_bytes` falls to the new cap
3578    /// on the shrink itself, not on eventual same-class reuse.
3579    #[mz_ore::test]
3580    fn budget_shrink_trims_warm_pool() {
3581        // Budget 8 MiB: warm cap 1 MiB, so four 64 KiB frees all park warm.
3582        let pool = test_pool(8 << 20);
3583        let handles: Vec<_> = (0..4)
3584            .map(|seed| insert(&pool, &mut payload(SMALL, 950 + seed)))
3585            .collect();
3586        drop(handles);
3587        assert_eq!(pool.stats().warm_bytes, 4 * (64 << 10));
3588
3589        // Budget 1 MiB: warm cap 128 KiB, so two of the four slots cool.
3590        pool.set_budget(1 << 20);
3591        assert_eq!(pool.stats().warm_bytes, 128 << 10);
3592    }
3593
3594    #[mz_ore::test]
3595    fn round_trip_resident() {
3596        let pool = test_pool(256 << 20);
3597        let orig = payload(1000, 1);
3598        let mut data = orig.clone();
3599        let capacity = data.capacity();
3600        let handle = insert(&pool, &mut data);
3601        assert!(data.is_empty());
3602        assert_eq!(data.capacity(), capacity, "insert preserves capacity");
3603        assert_eq!(handle.residency(), Residency::UnbackedResident);
3604        assert_eq!(read(&handle), orig);
3605        drop(handle);
3606        let stats = pool.stats();
3607        assert_eq!(stats.inserts, 1);
3608        assert_eq!(stats.frees, 1);
3609        assert_eq!(stats.resident_bytes, 0);
3610    }
3611
3612    #[mz_ore::test]
3613    fn take_reads_and_frees() {
3614        let pool = test_pool(256 << 20);
3615        let orig = payload(SMALL, 40);
3616        let handle = insert(&pool, &mut orig.clone());
3617        let mut out = Vec::new();
3618        handle.take(&mut out);
3619        assert_eq!(out, orig);
3620        let stats = pool.stats();
3621        assert_eq!(stats.frees, 1);
3622        assert_eq!(stats.writes_elided, 1, "a resident take never writes");
3623        assert_eq!(stats.resident_bytes, 0);
3624        assert_eq!(stats.live_chunks, 0);
3625    }
3626
3627    /// `prefetch` is safe wherever it lands: on a resident chunk (a no-op),
3628    /// on an evicted chunk (whose read then round-trips), and issued with no
3629    /// read following it. It never changes residency or resident bytes.
3630    #[mz_ore::test]
3631    fn prefetch_is_safe_in_every_state() {
3632        let pool = test_pool(256 << 20);
3633        let orig = payload(SMALL, 41);
3634        let handle = insert(&pool, &mut orig.clone());
3635        handle.prefetch();
3636        assert_eq!(handle.residency(), Residency::UnbackedResident);
3637        assert_eq!(read(&handle), orig);
3638        pool.evict(&handle);
3639        handle.prefetch();
3640        assert_eq!(handle.residency(), Residency::Evicted);
3641        assert_eq!(read(&handle), orig);
3642        // An advisory with no read behind it leaves nothing to clean up.
3643        let idle = insert(&pool, &mut payload(SMALL, 42));
3644        idle.prefetch();
3645        drop(idle);
3646        drop(handle);
3647        assert_eq!(pool.stats().resident_bytes, 0);
3648    }
3649
3650    /// Reading an evicted chunk decompresses its extent straight into the
3651    /// caller's buffer and leaves the chunk evicted. Free slots are poisoned
3652    /// first, so a read passing stale slot memory through (the macOS
3653    /// `MADV_DONTNEED` hazard) would fail the content check.
3654    #[mz_ore::test]
3655    fn evict_then_read_preserves_contents() {
3656        let pool = test_pool(256 << 20);
3657        let orig = payload(SMALL, 2);
3658        let handle = insert(&pool, &mut orig.clone());
3659        pool.evict(&handle);
3660        assert_eq!(handle.residency(), Residency::Evicted);
3661        let stats = pool.stats();
3662        assert_eq!(stats.evictions_compress, 1);
3663        assert_eq!(stats.resident_bytes, 0);
3664        assert!(stats.extent_bytes_written > 0);
3665        pool.poison_free_slots();
3666        assert_eq!(read(&handle), orig);
3667        assert_eq!(handle.residency(), Residency::Evicted);
3668        assert_eq!(pool.stats().resident_bytes, 0, "reads copy out");
3669    }
3670
3671    /// An admitting read of an evicted chunk with budget headroom re-admits
3672    /// it: contents round-trip, the chunk lands `BackedResident` with its
3673    /// extent kept, and later reads serve from the slot without touching
3674    /// the extent.
3675    #[mz_ore::test]
3676    fn admit_from_free_budget_backs_the_chunk() {
3677        let pool = test_pool(256 << 20);
3678        let orig = payload(SMALL, 70);
3679        let handle = insert(&pool, &mut orig.clone());
3680        pool.evict(&handle);
3681        assert_eq!(handle.residency(), Residency::Evicted);
3682        assert_eq!(pool.stats().resident_bytes, 0);
3683
3684        pool.poison_free_slots();
3685        assert_eq!(read_admit(&handle), orig);
3686        assert_eq!(handle.residency(), Residency::BackedResident);
3687        let stats = pool.stats();
3688        assert_eq!(stats.admissions_budget, 1);
3689        assert_eq!(stats.admissions_steal, 0);
3690        assert_eq!(stats.admissions_denied, 0);
3691        assert_eq!(stats.resident_bytes, 64 << 10);
3692
3693        // Later reads serve from the slot and never touch the extent: a
3694        // decompress would revive its pages and move the revival and
3695        // pageout counters.
3696        let pageouts = stats.extent_pageouts;
3697        let extent_resident = stats.extent_resident_bytes;
3698        assert_eq!(read(&handle), orig);
3699        assert_eq!(handle.residency(), Residency::BackedResident);
3700        let stats = pool.stats();
3701        assert_eq!(stats.extent_pageouts, pageouts);
3702        assert_eq!(stats.extent_resident_bytes, extent_resident);
3703
3704        // The kept extent pre-pays the next eviction, and round-trips.
3705        pool.evict(&handle);
3706        assert_eq!(handle.residency(), Residency::Evicted);
3707        let stats = pool.stats();
3708        assert_eq!(stats.evictions_cheap, 1);
3709        assert_eq!(stats.evictions_compress, 1, "admission wrote no extent");
3710        pool.poison_free_slots();
3711        assert_eq!(read(&handle), orig);
3712        drop(handle);
3713        assert_eq!(pool.stats().resident_bytes, 0);
3714    }
3715
3716    /// With the budget pinned full and a clean backed victim of the same
3717    /// class, an admitting read steals the victim's slot: the victim is
3718    /// evicted with zero I/O and its extent intact, the admitted chunk
3719    /// lands `BackedResident`, and resident bytes, warm bytes, and the
3720    /// compression and pageout counters are all unchanged.
3721    #[mz_ore::test]
3722    fn admit_steals_clean_victim_slot() {
3723        let pool = test_pool(256 << 20);
3724        pool.set_rss_target(1 << 30);
3725        let victim_orig = payload(SMALL, 71);
3726        let target_orig = payload(SMALL, 72);
3727        let victim = insert(&pool, &mut victim_orig.clone());
3728        let target = insert(&pool, &mut target_orig.clone());
3729        pool.evict(&target);
3730        assert!(pool.back_step(), "victim is backable");
3731        assert_eq!(victim.residency(), Residency::BackedResident);
3732        // The budget now holds exactly the victim: no admission headroom.
3733        pool.set_budget(64 << 10);
3734        assert_eq!(victim.residency(), Residency::BackedResident);
3735        let before = pool.stats();
3736
3737        assert_eq!(read_admit(&target), target_orig);
3738        assert_eq!(target.residency(), Residency::BackedResident);
3739        assert_eq!(victim.residency(), Residency::Evicted);
3740        let after = pool.stats();
3741        assert_eq!(after.admissions_steal, 1);
3742        assert_eq!(after.admissions_budget, 0);
3743        assert_eq!(after.admissions_denied, 0);
3744        assert_eq!(
3745            after.resident_bytes, before.resident_bytes,
3746            "same class, same bytes",
3747        );
3748        assert_eq!(
3749            after.evictions_compress, before.evictions_compress,
3750            "no compression",
3751        );
3752        assert_eq!(
3753            after.evictions_cheap, before.evictions_cheap,
3754            "a steal is not an enforcement eviction",
3755        );
3756        assert_eq!(after.extent_bytes_written, before.extent_bytes_written);
3757        assert_eq!(after.extent_pageouts, 0, "no pageout");
3758        assert_eq!(
3759            after.warm_bytes, before.warm_bytes,
3760            "the stolen slot skipped the free list",
3761        );
3762        assert_eq!(after.warm_reuses, before.warm_reuses);
3763
3764        // The victim's extent is intact: its old slot now holds the
3765        // admitted chunk's bytes, so a correct read must come from the
3766        // extent.
3767        assert_eq!(read(&victim), victim_orig);
3768        assert_eq!(victim.residency(), Residency::Evicted);
3769
3770        drop(victim);
3771        drop(target);
3772        let stats = pool.stats();
3773        assert_eq!(stats.resident_bytes, 0);
3774        assert_eq!(stats.extent_resident_bytes, 0);
3775    }
3776
3777    /// With the budget full and every candidate touched, the admitting read
3778    /// still returns correct data, the chunk stays evicted, and the denial
3779    /// counter increments.
3780    #[mz_ore::test]
3781    fn admit_denied_when_victims_touched() {
3782        let pool = test_pool(256 << 20);
3783        let victim_orig = payload(SMALL, 73);
3784        let target_orig = payload(SMALL, 74);
3785        let victim = insert(&pool, &mut victim_orig.clone());
3786        let target = insert(&pool, &mut target_orig.clone());
3787        pool.evict(&target);
3788        assert!(pool.back_step());
3789        // Reading the victim sets its second-chance bit, disqualifying it.
3790        assert_eq!(read(&victim), victim_orig);
3791        pool.set_budget(64 << 10);
3792        let resident = pool.stats().resident_bytes;
3793
3794        assert_eq!(read_admit(&target), target_orig);
3795        assert_eq!(target.residency(), Residency::Evicted);
3796        assert_eq!(victim.residency(), Residency::BackedResident);
3797        let stats = pool.stats();
3798        assert_eq!(stats.admissions_denied, 1);
3799        assert_eq!(stats.admissions_budget, 0);
3800        assert_eq!(stats.admissions_steal, 0);
3801        assert_eq!(stats.resident_bytes, resident);
3802    }
3803
3804    /// An unbacked resident candidate is never stolen from: evicting it
3805    /// would require the compression that admission forbids.
3806    #[mz_ore::test]
3807    fn admit_denied_when_victims_unbacked() {
3808        let pool = test_pool(256 << 20);
3809        let victim = insert(&pool, &mut payload(SMALL, 75));
3810        let target_orig = payload(SMALL, 76);
3811        let target = insert(&pool, &mut target_orig.clone());
3812        pool.evict(&target);
3813        pool.set_budget(64 << 10);
3814        assert_eq!(read_admit(&target), target_orig);
3815        assert_eq!(target.residency(), Residency::Evicted);
3816        assert_eq!(victim.residency(), Residency::UnbackedResident);
3817        assert_eq!(pool.stats().admissions_denied, 1);
3818    }
3819
3820    /// A clean backed victim of a different size class is never stolen
3821    /// from: slot reuse in place requires the classes to match.
3822    #[mz_ore::test]
3823    fn admit_denied_when_victims_wrong_class() {
3824        let pool = test_pool(256 << 20);
3825        // The victim fills the 128 KiB class; the target lives in the
3826        // 64 KiB one.
3827        let victim = insert(&pool, &mut payload(2 * SMALL, 77));
3828        let target_orig = payload(SMALL, 78);
3829        let target = insert(&pool, &mut target_orig.clone());
3830        pool.evict(&target);
3831        assert!(pool.back_step());
3832        assert_eq!(victim.residency(), Residency::BackedResident);
3833        // The budget holds exactly the victim: no headroom for the target.
3834        pool.set_budget(128 << 10);
3835        assert_eq!(read_admit(&target), target_orig);
3836        assert_eq!(target.residency(), Residency::Evicted);
3837        assert_eq!(victim.residency(), Residency::BackedResident);
3838        assert_eq!(pool.stats().admissions_denied, 1);
3839    }
3840
3841    #[mz_ore::test]
3842    fn plain_read_and_take_never_admit() {
3843        let pool = test_pool(256 << 20);
3844        let orig = payload(SMALL, 79);
3845        let handle = insert(&pool, &mut orig.clone());
3846        pool.evict(&handle);
3847        assert_eq!(read(&handle), orig);
3848        assert_eq!(handle.residency(), Residency::Evicted);
3849        assert_eq!(pool.stats().resident_bytes, 0);
3850        let mut out = Vec::new();
3851        handle.take(&mut out);
3852        assert_eq!(out, orig);
3853        let stats = pool.stats();
3854        assert_eq!(stats.admissions_budget, 0);
3855        assert_eq!(stats.admissions_steal, 0);
3856        assert_eq!(stats.admissions_denied, 0);
3857        assert_eq!(stats.resident_bytes, 0);
3858        assert_eq!(stats.frees, 1);
3859    }
3860
3861    /// A steal settles the ledger with the payload difference: a shrinking
3862    /// steal always proceeds, while a steal that would grow resident bytes
3863    /// past the budget is denied and leaves the victim untouched.
3864    #[mz_ore::test]
3865    fn steal_admission_charges_the_budget() {
3866        // Shrinking steal: the victim is larger than the admitted payload,
3867        // so the steal lowers resident bytes and always may proceed.
3868        let pool = test_pool(256 << 20);
3869        let victim_orig = payload(SMALL, 84);
3870        let victim = insert(&pool, &mut victim_orig.clone());
3871        assert!(pool.back_step(), "victim backs");
3872        let small_orig = payload(SMALL / 2, 85);
3873        let handle = insert(&pool, &mut small_orig.clone());
3874        pool.evict(&handle);
3875        pool.set_budget(64 << 10);
3876        assert_eq!(read_admit(&handle), small_orig);
3877        let stats = pool.stats();
3878        assert_eq!(stats.admissions_steal, 1, "no headroom, so the read steals");
3879        assert_eq!(stats.resident_bytes, u64::cast_from(SMALL / 2 * WORD_BYTES));
3880        assert_eq!(victim.residency(), Residency::Evicted);
3881        assert_eq!(read(&victim), victim_orig, "victim serves from its extent");
3882
3883        // Growing steal: the admitted payload is larger than the only
3884        // victim, and the growth does not fit the budget, so the admission
3885        // is denied and the victim is left untouched.
3886        let pool = test_pool(256 << 20);
3887        let big_orig = payload(SMALL, 86);
3888        let big = insert(&pool, &mut big_orig.clone());
3889        pool.evict(&big);
3890        let small_victim = insert(&pool, &mut payload(SMALL / 2, 87));
3891        assert!(pool.back_step(), "victim backs");
3892        pool.set_budget(32 << 10);
3893        assert_eq!(read_admit(&big), big_orig);
3894        let stats = pool.stats();
3895        assert_eq!(stats.admissions_denied, 1, "growth exceeds the budget");
3896        assert_eq!(stats.admissions_steal, 0);
3897        assert_eq!(big.residency(), Residency::Evicted);
3898        assert_eq!(small_victim.residency(), Residency::BackedResident);
3899    }
3900
3901    /// Admission of a chunk whose extent was pushed to the device: the read
3902    /// revives the extent into the acquired slot and re-counts it.
3903    #[mz_ore::test]
3904    fn admission_revives_paged_out_extent() {
3905        let pool = test_pool(256 << 20);
3906        let orig = payload(SMALL, 88);
3907        let handle = insert(&pool, &mut orig.clone());
3908        pool.evict(&handle);
3909        assert_eq!(
3910            pool.stats().extent_resident_bytes,
3911            0,
3912            "zero RSS target pages the extent out on eviction",
3913        );
3914        // Raise the target so the read's own tier enforcement does not
3915        // page the revived extent straight back out.
3916        pool.set_rss_target(1 << 30);
3917        assert_eq!(read_admit(&handle), orig);
3918        assert_eq!(handle.residency(), Residency::BackedResident);
3919        let stats = pool.stats();
3920        assert_eq!(stats.admissions_budget, 1);
3921        assert!(stats.extent_resident_bytes > 0, "revived and re-counted");
3922    }
3923
3924    /// An admitting read of a chunk that is not evicted is a plain read:
3925    /// no admission counter moves and no state changes.
3926    #[mz_ore::test]
3927    fn admit_is_plain_read_on_non_evicted_chunks() {
3928        let pool = test_pool(256 << 20);
3929        let orig = payload(SMALL, 89);
3930        let resident = insert(&pool, &mut orig.clone());
3931        assert_eq!(read_admit(&resident), orig);
3932        assert_eq!(resident.residency(), Residency::UnbackedResident);
3933        let words = SIZE_CLASSES[SIZE_CLASSES.len() - 1] / WORD_BYTES + 1;
3934        let oversize_orig = payload(words, 90);
3935        let oversize = insert(&pool, &mut oversize_orig.clone());
3936        assert_eq!(read_admit(&oversize), oversize_orig);
3937        let empty = insert(&pool, &mut Vec::new());
3938        assert!(read_admit(&empty).is_empty());
3939        let stats = pool.stats();
3940        assert_eq!(stats.admissions_budget, 0);
3941        assert_eq!(stats.admissions_steal, 0);
3942        assert_eq!(stats.admissions_denied, 0);
3943    }
3944
3945    /// Concurrent admitting reads with no budget headroom: every admission
3946    /// must go through the steal path, racing steals against each other on
3947    /// the same victims (the pool's only two-chunk lock edge). Contents are
3948    /// asserted on every read.
3949    #[mz_ore::test]
3950    #[cfg_attr(miri, ignore)] // too slow
3951    fn concurrent_admits_exercise_the_steal_path() {
3952        const CHUNKS: u64 = 8;
3953        let pool = test_pool(usize::MAX);
3954        let origs: Vec<_> = (0..CHUNKS).map(|seed| payload(SMALL, 900 + seed)).collect();
3955        let evicted: Arc<Vec<(Vec<u64>, ChunkHandle)>> = Arc::new(
3956            origs
3957                .iter()
3958                .map(|orig| {
3959                    let handle = insert(&pool, &mut orig.clone());
3960                    pool.evict(&handle);
3961                    (orig.clone(), handle)
3962                })
3963                .collect(),
3964        );
3965        let mut victims = Vec::new();
3966        for seed in 0..CHUNKS {
3967            victims.push(insert(&pool, &mut payload(SMALL, 950 + seed)));
3968            assert!(pool.back_step(), "victim backs");
3969        }
3970        // Exactly the victims' bytes: no free headroom, so every admission
3971        // steals or is denied.
3972        pool.set_budget(usize::cast_from(CHUNKS) * (64 << 10));
3973        let threads: Vec<_> = (0..2u64)
3974            .map(|t| {
3975                let evicted = Arc::clone(&evicted);
3976                std::thread::spawn(move || {
3977                    for round in 0..CHUNKS {
3978                        let (orig, handle) = &evicted[usize::cast_from((t + round) % CHUNKS)];
3979                        let mut out = Vec::new();
3980                        handle.read_into_admit(&mut out);
3981                        assert_eq!(&out, orig);
3982                    }
3983                })
3984            })
3985            .collect();
3986        for thread in threads {
3987            thread.join().expect("admitting thread panicked");
3988        }
3989        let stats = pool.stats();
3990        assert!(stats.admissions_steal > 0, "no headroom forces steals");
3991        assert!(
3992            stats.resident_bytes <= u64::cast_from(usize::cast_from(CHUNKS) * (64 << 10)),
3993            "steals never grow resident bytes past the budget",
3994        );
3995        // The victims were held live as steal targets; a steal leaves its
3996        // victim evicted, so at least one is evicted here.
3997        let stolen = victims
3998            .iter()
3999            .filter(|v| v.residency() == Residency::Evicted)
4000            .count();
4001        assert!(stolen > 0, "a steal evicts its victim");
4002    }
4003
4004    /// A re-admitted chunk keeps its insert-time depth: under budget
4005    /// pressure it is evicted from its own deeper band before a younger
4006    /// band-0 chunk, which a re-admission into band 0 would have inverted.
4007    #[mz_ore::test]
4008    fn admitted_chunk_keeps_its_depth() {
4009        let pool = test_pool(256 << 20);
4010        let deep_orig = payload(SMALL, 80);
4011        let deep = insert_at_depth(&pool, 2, &mut deep_orig.clone());
4012        let young = insert(&pool, &mut payload(SMALL, 81));
4013        pool.evict(&deep);
4014        assert_eq!(read_admit(&deep), deep_orig);
4015        assert_eq!(deep.residency(), Residency::BackedResident);
4016        assert_eq!(pool.stats().admissions_budget, 1);
4017
4018        // Budget of one chunk: enforcement visits the deep band first.
4019        pool.set_budget(64 << 10);
4020        assert_eq!(deep.residency(), Residency::Evicted);
4021        assert_eq!(young.residency(), Residency::UnbackedResident);
4022        assert_eq!(
4023            pool.stats().evictions_cheap,
4024            1,
4025            "the extent kept through admission pre-paid the eviction",
4026        );
4027    }
4028
4029    /// Admitting reads racing enforcement, opposing steals, and frees:
4030    /// contents stay correct, contended steals degrade to skips, and the
4031    /// accounting identity settles to zero.
4032    #[mz_ore::test]
4033    #[cfg_attr(miri, ignore)] // too slow
4034    fn concurrent_admits_race_cleanly() {
4035        let pool = test_pool(64 << 10);
4036        let per_thread = rounds(50, 3);
4037        let threads: Vec<_> = (0..4u64)
4038            .map(|t| {
4039                let pool = pool.clone();
4040                std::thread::spawn(move || {
4041                    let mut out = Vec::new();
4042                    for round in 0..per_thread {
4043                        let orig = payload(SMALL, t * 1000 + round);
4044                        let handle = insert(&pool, &mut orig.clone());
4045                        pool.evict(&handle);
4046                        handle.read_into_admit(&mut out);
4047                        assert_eq!(out, orig);
4048                        handle.read_into_admit(&mut out);
4049                        assert_eq!(out, orig);
4050                        assert_eq!(read(&handle), orig);
4051                    }
4052                })
4053            })
4054            .collect();
4055        for thread in threads {
4056            thread.join().expect("worker thread panicked");
4057        }
4058        let stats = pool.stats();
4059        assert_eq!(stats.inserts, 4 * per_thread);
4060        assert_eq!(stats.frees, 4 * per_thread);
4061        assert_eq!(stats.resident_bytes, 0);
4062        assert_eq!(stats.extent_resident_bytes, 0);
4063    }
4064
4065    /// Slots are scoped to residency: eviction releases the slot, so a
4066    /// capacity holding exactly one chunk can serve any number of chunks one
4067    /// at a time, and reads of evicted chunks need no slot at all.
4068    #[mz_ore::test]
4069    fn eviction_releases_the_slot() {
4070        // One 64 KiB slot per class.
4071        let pool = Pool::with_class_capacity(64 << 10).expect("pool creation");
4072        let a = insert(&pool, &mut payload(SMALL, 6));
4073        pool.evict(&a);
4074        // The class's only slot is free again: a second chunk fits without
4075        // falling back to the heap.
4076        let b = insert(&pool, &mut payload(SMALL, 7));
4077        assert_eq!(b.residency(), Residency::UnbackedResident);
4078        assert_eq!(pool.stats().slot_exhausted_fallbacks, 0);
4079        // Reading `a` decompresses straight from its extent while `b` holds
4080        // the class's only slot: copy-out allocates nothing.
4081        assert_eq!(read(&a), payload(SMALL, 6));
4082        assert_eq!(a.residency(), Residency::Evicted);
4083        assert_eq!(read(&b), payload(SMALL, 7));
4084    }
4085
4086    /// The eviction queue holds resident chunks only: an enforcement pass
4087    /// drops entries for evicted chunks, and reads never re-add them, so the
4088    /// scan each insert pays stays proportional to the resident set rather
4089    /// than every chunk ever evicted.
4090    #[mz_ore::test]
4091    fn queue_holds_resident_chunks_only() {
4092        let pool = test_pool(128 << 10);
4093        let mut handles = Vec::new();
4094        for seed in 0..8 {
4095            handles.push(insert(&pool, &mut payload(SMALL, 800 + seed)));
4096        }
4097        // Budget pressure evicted ~6 of 8; one more pass visits the evicted
4098        // entries and drops them (their first visit performed the eviction
4099        // and dropped them already, but second-chance survivors may linger).
4100        pool.enforce_budget();
4101        let resident = handles
4102            .iter()
4103            .filter(|h| h.residency() != Residency::Evicted)
4104            .count();
4105        assert!(
4106            pool.queue_len() <= resident + 1,
4107            "queue ({}) tracks the resident set ({resident}), not all 8 live chunks",
4108            pool.queue_len(),
4109        );
4110        // Reading an evicted chunk copies out of its extent and does not
4111        // re-enqueue it: the queue keeps tracking the resident set.
4112        let evicted = handles
4113            .iter()
4114            .find(|h| h.residency() == Residency::Evicted)
4115            .expect("something was evicted");
4116        let before = pool.queue_len();
4117        assert_eq!(read(evicted).len(), SMALL);
4118        assert_eq!(evicted.residency(), Residency::Evicted);
4119        assert_eq!(pool.queue_len(), before, "reads leave the queue alone");
4120    }
4121
4122    #[mz_ore::test]
4123    fn dead_data_is_never_written() {
4124        let pool = test_pool(256 << 20);
4125        let handle = insert(&pool, &mut payload(SMALL, 7));
4126        drop(handle);
4127        let stats = pool.stats();
4128        assert_eq!(stats.frees, 1);
4129        assert_eq!(stats.writes_elided, 1);
4130        assert_eq!(stats.extent_bytes_written, 0);
4131        assert_eq!(stats.resident_bytes, 0);
4132    }
4133
4134    #[mz_ore::test]
4135    fn budget_is_enforced_on_insert() {
4136        let budget = 128 << 10;
4137        let pool = test_pool(budget);
4138        let mut handles = Vec::new();
4139        for seed in 0..8 {
4140            handles.push(insert(&pool, &mut payload(SMALL, 100 + seed)));
4141        }
4142        let stats = pool.stats();
4143        assert!(
4144            stats.resident_bytes <= u64::cast_from(budget),
4145            "resident {} exceeds budget {}",
4146            stats.resident_bytes,
4147            budget,
4148        );
4149        assert!(stats.evictions_compress >= 6);
4150        let resident = handles
4151            .iter()
4152            .filter(|h| {
4153                matches!(
4154                    h.residency(),
4155                    Residency::UnbackedResident | Residency::BackedResident
4156                )
4157            })
4158            .count();
4159        assert_eq!(resident, 2, "budget holds exactly two small chunks");
4160    }
4161
4162    /// Budget enforcement is single-flight: an insert that trips it while a
4163    /// pass holds the `enforcing` guard bails on `WouldBlock`, trusting that
4164    /// pass. If the holder is already past its final `resident_bytes` read, the
4165    /// bailed insert's bytes are neither read by the holder nor enforced by the
4166    /// bailer, and no later insert re-trips enforcement, so the pool stays over
4167    /// budget. The fix re-runs the pass while any caller was turned away.
4168    ///
4169    /// The test hook freezes the holder's pass in that window to make the race
4170    /// deterministic: the holder parks having found the budget satisfied, the
4171    /// main thread inserts over budget and is turned away, then the holder
4172    /// resumes. The `gate` is used for both rendezvous.
4173    #[mz_ore::test]
4174    fn racing_insert_is_not_dropped_by_budget_single_flight() {
4175        // Budget for exactly one small chunk.
4176        let budget = 64 << 10;
4177        let pool = test_pool(budget);
4178        let gate = std::sync::Arc::new(std::sync::Barrier::new(2));
4179
4180        let holder = {
4181            let pool = pool.clone();
4182            let gate = std::sync::Arc::clone(&gate);
4183            std::thread::spawn(move || -> ChunkHandle {
4184                ENFORCE_BUDGET_HOOK.with(|cell| {
4185                    *cell.borrow_mut() = Some(Box::new(move || {
4186                        gate.wait(); // parked, holding the guard
4187                        gate.wait(); // resume once the race is done
4188                    }));
4189                });
4190                // At budget: the pass finds it satisfied and parks at the hook.
4191                insert(&pool, &mut payload(SMALL, 1))
4192            })
4193        };
4194
4195        gate.wait(); // holder is parked in enforcement, holding the guard
4196        // Push over budget; this insert is turned away by the held guard.
4197        let _over = insert(&pool, &mut payload(SMALL, 2));
4198        gate.wait(); // let the holder resume and release the guard
4199        // Kept alive past the assert: freeing it would drop its bytes and mask
4200        // the overshoot.
4201        let _held = holder.join().expect("holder panicked");
4202
4203        // Nothing re-trips enforcement, so the pool must not be left over budget.
4204        let resident = pool.stats().resident_bytes;
4205        assert!(
4206            resident <= u64::cast_from(budget),
4207            "resident {resident} exceeds budget {budget}: racing insert escaped enforcement",
4208        );
4209    }
4210
4211    #[mz_ore::test]
4212    fn insertion_debt_is_bounded_during_enforcement() {
4213        let budget = 2 * SMALL * WORD_BYTES;
4214        let pool = test_pool(budget);
4215        let guard = pool.0.enforcing.lock().expect("enforcement lock");
4216        let mut handles = Vec::new();
4217        for seed in 0..16 {
4218            handles.push(insert(&pool, &mut payload(SMALL, seed)));
4219            assert!(
4220                pool.stats().resident_bytes <= u64::cast_from(budget + SMALL * WORD_BYTES),
4221                "an occupied enforcer must not allow unlimited insertion debt",
4222            );
4223        }
4224        drop(guard);
4225        for (seed, handle) in handles.iter().enumerate() {
4226            assert_eq!(read(handle), payload(SMALL, u64::cast_from(seed)));
4227        }
4228        drop(handles);
4229        assert_eq!(pool.stats().resident_bytes, 0);
4230        assert_eq!(pool.stats().live_chunks, 0);
4231        assert_eq!(pool.stats().extent_resident_bytes, 0);
4232    }
4233
4234    #[mz_ore::test]
4235    fn admission_reserves_before_concurrent_fills() {
4236        let budget = 2 * SMALL * WORD_BYTES;
4237        let pool = test_pool(budget);
4238        let guard = pool.0.enforcing.lock().expect("enforcement lock");
4239        let gate = Arc::new(std::sync::Barrier::new(9));
4240        let threads: Vec<_> = (0..8u64)
4241            .map(|seed| {
4242                let pool = pool.clone();
4243                let gate = Arc::clone(&gate);
4244                std::thread::spawn(move || {
4245                    pool.insert_with(SMALL, ChunkHints::default(), &TEST_CODEC, |dst| {
4246                        gate.wait();
4247                        gate.wait();
4248                        dst.copy_from_slice(&payload(SMALL, seed));
4249                    })
4250                })
4251            })
4252            .collect();
4253        gate.wait();
4254        let reserved = pool.stats().resident_bytes;
4255        // Release every producer even if the assertion fails.
4256        gate.wait();
4257        let handles: Vec<_> = threads
4258            .into_iter()
4259            .map(|t| t.join().expect("producer panicked"))
4260            .collect();
4261        drop(guard);
4262        assert!(reserved <= u64::cast_from(budget + SMALL * WORD_BYTES));
4263        assert!(pool.stats().direct_extent_inserts > 0);
4264        for (seed, handle) in handles.iter().enumerate() {
4265            assert_eq!(read(handle), payload(SMALL, u64::cast_from(seed)));
4266        }
4267        drop(handles);
4268        assert_eq!(pool.stats().resident_bytes, 0);
4269        assert_eq!(pool.stats().live_chunks, 0);
4270    }
4271
4272    #[mz_ore::test]
4273    fn set_budget_retunes_in_place() {
4274        let pool = test_pool(usize::MAX);
4275        let mut handles = Vec::new();
4276        for seed in 0..8 {
4277            handles.push(insert(&pool, &mut payload(SMALL, 200 + seed)));
4278        }
4279        assert_eq!(pool.stats().evictions_compress, 0);
4280
4281        // Shrinking the budget evicts immediately.
4282        pool.set_budget(128 << 10);
4283        let stats = pool.stats();
4284        assert!(stats.resident_bytes <= 128 << 10);
4285        assert!(stats.evictions_compress >= 6);
4286
4287        // Growing it leaves headroom: a fresh insert stays resident.
4288        pool.set_budget(usize::MAX);
4289        let h = insert(&pool, &mut payload(SMALL, 300));
4290        assert_eq!(h.residency(), Residency::UnbackedResident);
4291        for h in &handles {
4292            assert_eq!(read(h).len(), SMALL);
4293        }
4294    }
4295
4296    #[mz_ore::test]
4297    fn second_chance_prefers_untouched_victims() {
4298        // Budget holds one and a half small chunks.
4299        let pool = test_pool((64 << 10) + (32 << 10));
4300        let orig_a = payload(SMALL, 8);
4301        let handle_a = insert(&pool, &mut orig_a.clone());
4302        assert_eq!(read(&handle_a), orig_a);
4303        // Inserting B overflows the budget; A is older but touched, so the
4304        // enforcer gives it a second chance and evicts untouched B instead.
4305        let handle_b = insert(&pool, &mut payload(SMALL, 9));
4306        assert_eq!(handle_a.residency(), Residency::UnbackedResident);
4307        assert_eq!(handle_b.residency(), Residency::Evicted);
4308    }
4309
4310    /// Depth-hinted chunks are evicted before younger ones: the deep chunk
4311    /// loses even though the young chunk is older and both are untouched
4312    /// (plain FIFO would have evicted the older, young one). Also exercises
4313    /// band clamping: depths beyond the last band share it.
4314    #[mz_ore::test]
4315    fn eviction_prefers_deeper_chunks() {
4316        // Budget of one small chunk.
4317        let pool = test_pool(64 << 10);
4318        let young = insert(&pool, &mut payload(SMALL, 900));
4319        let deep = insert_at_depth(&pool, 255, &mut payload(SMALL, 901));
4320        assert_eq!(young.residency(), Residency::UnbackedResident);
4321        assert_eq!(deep.residency(), Residency::Evicted);
4322    }
4323
4324    /// Eager backing visits deeper chunks first, mirroring eviction order,
4325    /// so the chunks evicted first are the ones already backed.
4326    #[mz_ore::test]
4327    fn backing_prefers_deeper_chunks() {
4328        let pool = test_pool(256 << 20);
4329        let young = insert(&pool, &mut payload(SMALL, 902));
4330        let deep = insert_at_depth(&pool, 2, &mut payload(SMALL, 903));
4331        assert!(pool.back_step());
4332        assert_eq!(deep.residency(), Residency::BackedResident);
4333        assert_eq!(young.residency(), Residency::UnbackedResident);
4334        assert!(pool.back_step());
4335        assert_eq!(young.residency(), Residency::BackedResident);
4336    }
4337
4338    #[mz_ore::test]
4339    fn empty_insert_consumes_no_slot() {
4340        let pool = test_pool(256 << 20);
4341        let mut data = Vec::new();
4342        let handle = insert(&pool, &mut data);
4343        assert_eq!(handle.size_class_bytes(), None);
4344        assert!(read(&handle).is_empty());
4345        // Reads clear the destination even for empty chunks.
4346        let mut out = vec![1u64, 2, 3];
4347        handle.read_into(&mut out);
4348        assert!(out.is_empty());
4349        drop(handle);
4350        let stats = pool.stats();
4351        assert_eq!(stats.resident_bytes, 0);
4352        assert_eq!(stats.writes_elided, 0);
4353    }
4354
4355    #[mz_ore::test]
4356    fn oversize_round_trips() {
4357        let pool = test_pool(256 << 20);
4358        let words = SIZE_CLASSES[SIZE_CLASSES.len() - 1] / WORD_BYTES + 1;
4359        let orig = payload(words, 10);
4360        let handle = insert(&pool, &mut orig.clone());
4361        assert_eq!(handle.residency(), Residency::Oversize);
4362        assert_eq!(handle.size_class_bytes(), None);
4363        let stats = pool.stats();
4364        assert_eq!(stats.oversize_bytes, u64::cast_from(words * WORD_BYTES));
4365        // The payload outgrew the largest class, and no class was exhausted.
4366        assert_eq!(stats.oversize_payloads, 1);
4367        assert_eq!(stats.slot_exhausted_fallbacks, 0);
4368        // Explicit eviction and budget enforcement leave oversize chunks
4369        // resident.
4370        pool.evict(&handle);
4371        pool.enforce_budget();
4372        assert_eq!(handle.residency(), Residency::Oversize);
4373        assert_eq!(read(&handle), orig);
4374        drop(handle);
4375        let stats = pool.stats();
4376        assert_eq!(stats.oversize_bytes, 0);
4377        assert_eq!(stats.resident_bytes, 0);
4378    }
4379
4380    #[mz_ore::test]
4381    fn payload_lands_in_smallest_fitting_class() {
4382        let pool = test_pool(256 << 20);
4383        let handle = insert(&pool, &mut payload((100 << 10) / WORD_BYTES, 11));
4384        assert_eq!(handle.size_class_bytes(), Some(128 << 10));
4385        let exact = insert(&pool, &mut payload(SMALL, 12));
4386        assert_eq!(exact.size_class_bytes(), Some(64 << 10));
4387    }
4388
4389    #[mz_ore::test]
4390    #[cfg_attr(miri, ignore)] // too slow
4391    fn multithreaded_smoke() {
4392        // Budget of one small chunk: four inserting threads keep the pool
4393        // over budget, so every insert's enforcement pass selects victims
4394        // owned by other threads, racing cross-thread eviction against
4395        // copy-out reads and frees.
4396        let pool = test_pool(64 << 10);
4397        let per_thread = rounds(50, 3);
4398        let threads: Vec<_> = (0..4u64)
4399            .map(|t| {
4400                let pool = pool.clone();
4401                std::thread::spawn(move || {
4402                    for round in 0..per_thread {
4403                        let seed = t * 1000 + round;
4404                        let orig = payload(SMALL, seed);
4405                        let handle = insert(&pool, &mut orig.clone());
4406                        pool.evict(&handle);
4407                        assert_eq!(read(&handle), orig);
4408                        // Enforcement racing reads must never corrupt them.
4409                        pool.enforce_budget();
4410                        assert_eq!(read(&handle), orig);
4411                        drop(handle);
4412                    }
4413                })
4414            })
4415            .collect();
4416        for thread in threads {
4417            thread.join().expect("worker thread panicked");
4418        }
4419        let stats = pool.stats();
4420        assert_eq!(stats.inserts, 4 * per_thread);
4421        assert_eq!(stats.frees, 4 * per_thread);
4422        assert_eq!(stats.resident_bytes, 0);
4423    }
4424
4425    #[mz_ore::test]
4426    #[cfg_attr(miri, ignore)] // too slow
4427    fn concurrent_read_enforce_churn() {
4428        // Races the three actors that can touch one chunk's slot: readers
4429        // copying shared chunks out and verifying them, an enforcer evicting
4430        // them (the zero budget makes every chunk a victim), and a churner
4431        // whose insert/free traffic turns the queue over. Contents are
4432        // asserted on every read, so an eviction or slot recycle racing a
4433        // copy-out shows up as corruption.
4434        let pool = test_pool(0);
4435        let shared: Arc<Vec<(Vec<u64>, ChunkHandle)>> = Arc::new(
4436            (0..4u64)
4437                .map(|seed| {
4438                    let orig = payload(SMALL, 600 + seed);
4439                    let handle = insert(&pool, &mut orig.clone());
4440                    (orig, handle)
4441                })
4442                .collect(),
4443        );
4444        let churn = rounds(300, 6);
4445        let mut threads = Vec::new();
4446        for t in 0..2u64 {
4447            let shared = Arc::clone(&shared);
4448            threads.push(std::thread::spawn(move || {
4449                for round in 0..churn {
4450                    let (orig, handle) = &shared[usize::cast_from((t + round) % 4)];
4451                    assert_eq!(&read(handle), orig);
4452                }
4453            }));
4454        }
4455        {
4456            let pool = pool.clone();
4457            threads.push(std::thread::spawn(move || {
4458                for _ in 0..2 * churn {
4459                    pool.enforce_budget();
4460                }
4461            }));
4462        }
4463        {
4464            let pool = pool.clone();
4465            threads.push(std::thread::spawn(move || {
4466                for round in 0..churn {
4467                    let orig = payload(SMALL, 700 + round);
4468                    let handle = insert(&pool, &mut orig.clone());
4469                    assert_eq!(read(&handle), orig);
4470                }
4471            }));
4472        }
4473        for thread in threads {
4474            thread.join().expect("worker thread panicked");
4475        }
4476        drop(shared);
4477        assert_eq!(pool.stats().resident_bytes, 0);
4478    }
4479
4480    /// Read-only traffic never raises resident bytes: every chunk starts
4481    /// evicted and is then read once, with no inserts in between. Reads copy
4482    /// out of the extents and leave every chunk evicted, so a seek-heavy
4483    /// phase costs no pool memory at all.
4484    #[mz_ore::test]
4485    fn reads_never_raise_resident_bytes() {
4486        let pool = test_pool(128 << 10);
4487        let origs: Vec<_> = (0..8u64).map(|seed| payload(SMALL, 300 + seed)).collect();
4488        let handles: Vec<_> = origs
4489            .iter()
4490            .map(|o| insert(&pool, &mut o.clone()))
4491            .collect();
4492        for handle in &handles {
4493            pool.evict(handle);
4494        }
4495        assert_eq!(pool.stats().resident_bytes, 0);
4496        for (index, handle) in handles.iter().enumerate() {
4497            assert_eq!(read(handle), origs[index]);
4498            assert_eq!(handle.residency(), Residency::Evicted);
4499            assert_eq!(pool.stats().resident_bytes, 0);
4500        }
4501    }
4502
4503    /// Evict-then-free churn under a generous RSS target: the compressed
4504    /// tier never crosses its cap, so enforcement never visits (and never
4505    /// drops) extent-queue entries, and pruning alone must keep the queue
4506    /// proportional to the live resident extents.
4507    #[mz_ore::test]
4508    #[cfg_attr(miri, ignore)] // too slow
4509    fn extent_queue_stays_bounded_under_cap() {
4510        let pool = test_pool(256 << 20);
4511        pool.set_rss_target(1 << 40);
4512        for seed in 0..rounds(1000, 48) {
4513            let handle = insert(&pool, &mut payload(SMALL, seed));
4514            pool.evict(&handle);
4515            drop(handle);
4516        }
4517        assert_eq!(pool.stats().extent_resident_bytes, 0);
4518        let len = pool.extent_queue_len();
4519        assert!(
4520            len <= 32,
4521            "extent queue holds {len} entries for zero resident extents",
4522        );
4523    }
4524
4525    /// A warm slot reused for a smaller payload round-trips: the tail
4526    /// release past the new payload must not disturb the payload itself,
4527    /// and the ledger credits exactly the payload.
4528    #[mz_ore::test]
4529    fn warm_reuse_with_smaller_payload_round_trips() {
4530        // Budget 8 MiB: warm cap = 1 MiB, so a 64 KiB slot parks warm.
4531        let pool = test_pool(8 << 20);
4532        let full = insert(&pool, &mut payload(SMALL, 60));
4533        drop(full);
4534        assert_eq!(pool.stats().warm_bytes, 64 << 10, "freed slot parks warm");
4535        // A payload of just over a page reuses the warm slot; the slot's
4536        // pages past it are released.
4537        let words = 4096 / WORD_BYTES + 1;
4538        let orig = payload(words, 61);
4539        let handle = insert(&pool, &mut orig.clone());
4540        let stats = pool.stats();
4541        assert_eq!(stats.warm_reuses, 1, "reused the warm slot");
4542        assert_eq!(stats.resident_bytes, u64::cast_from(words * WORD_BYTES));
4543        assert_eq!(read(&handle), orig);
4544        // Round-trips through the extent as well.
4545        pool.evict(&handle);
4546        pool.poison_free_slots();
4547        assert_eq!(read(&handle), orig);
4548        drop(handle);
4549        assert_eq!(pool.stats().resident_bytes, 0);
4550    }
4551
4552    /// Heap-backed chunks count as resident but can never be evicted, so
4553    /// the budget must not force slotted chunks out on their account: with
4554    /// unevictable bytes alone exceeding the budget, a slotted chunk that
4555    /// fits the budget stays resident.
4556    #[mz_ore::test]
4557    fn unevictable_bytes_do_not_force_eviction() {
4558        // One 64 KiB slot per class: the second and third inserts fall
4559        // back to the heap.
4560        let pool = Pool::with_class_capacity(64 << 10).expect("pool creation");
4561        pool.set_budget(64 << 10);
4562        let slotted = insert(&pool, &mut payload(SMALL, 91));
4563        let heap_a = insert(&pool, &mut payload(SMALL, 92));
4564        let heap_b = insert(&pool, &mut payload(SMALL, 93));
4565        assert_eq!(heap_a.residency(), Residency::Oversize);
4566        assert_eq!(heap_b.residency(), Residency::Oversize);
4567        let stats = pool.stats();
4568        assert!(stats.oversize_bytes > 64 << 10, "unevictable exceed budget");
4569        assert_eq!(slotted.residency(), Residency::UnbackedResident);
4570        assert_eq!(stats.evictions_compress, 0);
4571        assert_eq!(read(&slotted), payload(SMALL, 91));
4572        assert_eq!(read(&heap_a), payload(SMALL, 92));
4573    }
4574
4575    /// A slotless empty chunk survives an explicit evict with spill
4576    /// scheduling enabled: nothing is handed to the spill threads and the
4577    /// chunk stays readable.
4578    #[mz_ore::test]
4579    fn evict_of_empty_chunk_is_a_no_op() {
4580        let pool = test_pool(usize::MAX);
4581        pool.enable_spill_without_threads();
4582        let empty = insert(&pool, &mut Vec::new());
4583        pool.evict(&empty);
4584        assert_eq!(empty.residency(), Residency::UnbackedResident);
4585        assert!(!pool.spill_step(), "nothing was scheduled");
4586        assert_eq!(pool.stats().spill_scheduled, 0);
4587        assert!(read(&empty).is_empty());
4588    }
4589
4590    #[mz_ore::test]
4591    fn queue_stays_bounded_under_budget() {
4592        // Chunk churn that never exceeds the budget: the enforcer's eviction
4593        // loop never runs, so stale queue entries must be reclaimed by
4594        // pruning alone.
4595        let pool = test_pool(256 << 20);
4596        for seed in 0..rounds(1000, 48) {
4597            let handle = insert(&pool, &mut payload(SMALL, seed));
4598            drop(handle);
4599        }
4600        let len = pool.queue_len();
4601        assert!(len <= 32, "queue holds {len} entries for zero live chunks");
4602    }
4603
4604    #[mz_ore::test]
4605    fn budget_pass_charges_each_hand_off_once() {
4606        let pool = test_pool(usize::MAX);
4607        pool.enable_spill_without_threads();
4608        let handles: Vec<_> = (0..16)
4609            .map(|seed| insert(&pool, &mut payload(SMALL, 500 + seed)))
4610            .collect();
4611        // The first sweep hands off the 12 untouched chunks and spends the
4612        // 4 touched chunks' second chance. The second sweep revisits the
4613        // handed-off entries before it reaches the touched ones, and
4614        // charging those again would exhaust the quota first.
4615        for handle in &handles[12..] {
4616            let _ = read(handle);
4617        }
4618        pool.set_budget(0);
4619        for handle in &handles {
4620            assert_eq!(
4621                handle.residency(),
4622                Residency::WriteInFlight,
4623                "the pass hands off the whole overage it saw",
4624            );
4625        }
4626        while pool.spill_step() {}
4627        for (i, handle) in handles.iter().enumerate() {
4628            assert_eq!(read(handle), payload(SMALL, 500 + u64::cast_from(i)));
4629        }
4630    }
4631
4632    #[mz_ore::test]
4633    fn budget_pass_trims_the_tier_between_evictions() {
4634        const CHUNKS: u64 = 32;
4635        let pool = test_pool(256 << 20);
4636        pool.set_rss_target(1 << 30);
4637        let handles: Vec<_> = (0..CHUNKS)
4638            .map(|seed| insert(&pool, &mut payload(SMALL, 600 + seed)))
4639            .collect();
4640        // The largest extent a chunk evicts to, measured on a probe chunk.
4641        let extent = {
4642            let probe = insert(&pool, &mut payload(SMALL, 600));
4643            pool.evict(&probe);
4644            pool.stats().extent_resident_bytes
4645        };
4646        assert!(extent > 0);
4647        let cap = extent + extent / 2;
4648        // Spill threads that exist but take no hand-off: the backstop then
4649        // applies to inline passes, and nothing else pages out.
4650        pool.0.spill.threads.store(1, Ordering::Relaxed);
4651        // With a zero budget the slot budget, insertion slack, and warm cap
4652        // are all zero, so the whole target is tier capacity.
4653        pool.set_rss_target(usize::cast_from(cap));
4654
4655        // Shrinking the budget evicts every chunk inline in one pass. The
4656        // hook fires after the evicting pass and before the trim that
4657        // follows it, so it observes the tier the pass left behind.
4658        let after_pass = std::rc::Rc::new(std::cell::Cell::new(0));
4659        let observed = std::rc::Rc::clone(&after_pass);
4660        let inner = Arc::clone(&pool.0);
4661        ENFORCE_BUDGET_HOOK.with(|cell| {
4662            *cell.borrow_mut() = Some(Box::new(move || {
4663                observed.set(inner.counters.extent_resident_bytes.load(Ordering::Relaxed));
4664            }));
4665        });
4666        pool.set_budget(0);
4667        assert_eq!(pool.0.compressed_cap(), cap);
4668        assert!(
4669            after_pass.get() <= 2 * cap + extent,
4670            "tier {} after the pass exceeds the backstop {} plus one extent {}",
4671            after_pass.get(),
4672            2 * cap,
4673            extent,
4674        );
4675        for (i, handle) in handles.iter().enumerate() {
4676            assert_eq!(read(handle), payload(SMALL, 600 + u64::cast_from(i)));
4677        }
4678    }
4679
4680    #[mz_ore::test]
4681    fn spill_async_evict_round_trip() {
4682        let pool = test_pool(usize::MAX);
4683        pool.enable_spill_without_threads();
4684        let h = insert(&pool, &mut payload(SMALL, 400));
4685        pool.evict(&h);
4686        assert_eq!(h.residency(), Residency::WriteInFlight);
4687        // Readable while in flight: the slot is still populated, and the
4688        // copy-out coexists with the spill thread's compression read.
4689        assert_eq!(read(&h), payload(SMALL, 400));
4690        // Reads leave no trace, so the eviction commits.
4691        assert!(pool.spill_step());
4692        assert_eq!(h.residency(), Residency::Evicted);
4693        let stats = pool.stats();
4694        assert_eq!(stats.spill_scheduled, 1);
4695        assert_eq!(stats.evictions_compress, 1);
4696        pool.poison_free_slots();
4697        assert_eq!(read(&h), payload(SMALL, 400));
4698    }
4699
4700    #[mz_ore::test]
4701    fn spill_freed_while_queued_is_elided() {
4702        let pool = test_pool(usize::MAX);
4703        pool.enable_spill_without_threads();
4704        let h = insert(&pool, &mut payload(SMALL, 401));
4705        pool.evict(&h);
4706        assert_eq!(h.residency(), Residency::WriteInFlight);
4707        drop(h);
4708        assert!(pool.spill_step());
4709        let stats = pool.stats();
4710        assert_eq!(stats.spill_cancelled, 1);
4711        assert_eq!(stats.writes_elided, 1, "freed before compression: elided");
4712        assert_eq!(stats.extent_bytes_written, 0, "no extent was written");
4713        assert_eq!(stats.resident_bytes, 0, "slot accounting settled");
4714    }
4715
4716    #[mz_ore::test]
4717    fn spill_take_in_flight_cancels_write() {
4718        let pool = test_pool(usize::MAX);
4719        pool.enable_spill_without_threads();
4720        let orig = payload(SMALL, 402);
4721        let h = insert(&pool, &mut orig.clone());
4722        pool.evict(&h);
4723        assert_eq!(h.residency(), Residency::WriteInFlight);
4724        let mut out = Vec::new();
4725        h.take(&mut out);
4726        assert_eq!(out, orig);
4727        assert!(pool.spill_step());
4728        let stats = pool.stats();
4729        assert_eq!(stats.frees, 1);
4730        assert_eq!(stats.spill_cancelled, 1);
4731        assert_eq!(stats.writes_elided, 1, "taken before compression: elided");
4732        assert_eq!(stats.extent_bytes_written, 0, "no extent was written");
4733        assert_eq!(stats.resident_bytes, 0, "slot accounting settled");
4734        assert_eq!(stats.live_chunks, 0);
4735    }
4736
4737    #[mz_ore::test]
4738    #[cfg_attr(miri, ignore)] // too slow
4739    fn spill_threads_end_to_end() {
4740        let pool = test_pool(128 << 10);
4741        pool.set_spill_threads(2);
4742        let mut handles = Vec::new();
4743        for seed in 0..rounds(16, 6) {
4744            handles.push(insert(&pool, &mut payload(SMALL, 500 + seed)));
4745        }
4746        pool.quiesce_spill();
4747        let stats = pool.stats();
4748        assert!(
4749            stats.spill_scheduled > 0,
4750            "budget pressure should have scheduled spills",
4751        );
4752        for (i, h) in handles.iter().enumerate() {
4753            assert_eq!(read(h), payload(SMALL, 500 + u64::cast_from(i)));
4754        }
4755        pool.join_spill_threads();
4756    }
4757
4758    /// Races the `WriteInFlight` protocol in its true concurrent form:
4759    /// spill threads compress slots without the state lock while owner
4760    /// threads copy the same chunks out under it and drop chunks mid-flight
4761    /// (both cancellation windows). Contents are asserted on every read, so
4762    /// a compression or slot release racing a copy-out shows up as
4763    /// corruption; under Miri the aliasing itself is checked.
4764    #[mz_ore::test]
4765    fn spill_threads_race_reads_and_drops() {
4766        let pool = test_pool(usize::MAX);
4767        pool.set_spill_threads(2);
4768        let iters = rounds(50, 6);
4769        let mut threads = Vec::new();
4770        for t in 0..2u64 {
4771            let pool = pool.clone();
4772            threads.push(std::thread::spawn(move || {
4773                for round in 0..iters {
4774                    let orig = payload(SMALL, t * 10_000 + round);
4775                    let handle = insert(&pool, &mut orig.clone());
4776                    // Hands the chunk to the spill threads (`WriteInFlight`).
4777                    pool.evict(&handle);
4778                    // Copy-out read racing the unlocked compression read.
4779                    assert_eq!(read(&handle), orig);
4780                    if round % 2 == 0 {
4781                        // Free while queued or mid-compression: the
4782                        // cancellation windows own the deferred cleanup.
4783                        drop(handle);
4784                    } else {
4785                        assert_eq!(read(&handle), orig);
4786                    }
4787                }
4788            }));
4789        }
4790        for thread in threads {
4791            thread.join().expect("worker thread panicked");
4792        }
4793        pool.quiesce_spill();
4794        pool.join_spill_threads();
4795        assert_eq!(pool.stats().resident_bytes, 0);
4796    }
4797
4798    /// The identity codec stores the body verbatim: eviction and reads,
4799    /// whole and by range, reconstruct it unchanged.
4800    #[mz_ore::test]
4801    fn identity_codec_round_trips() {
4802        let pool = test_pool(usize::MAX);
4803        let want = payload(SMALL, 601);
4804        let h = pool.insert_with(SMALL, ChunkHints::default(), &IDENTITY_CODEC, |dst| {
4805            dst.copy_from_slice(&want);
4806        });
4807        assert_eq!(read(&h), want);
4808        pool.evict(&h);
4809        assert_eq!(read(&h), want, "round-trips through the extent");
4810        pool.evict(&h);
4811        let mut range = Vec::new();
4812        h.read_range_into(8..24, &mut range);
4813        assert_eq!(range, want[8..24], "range reads copy the range directly");
4814    }
4815
4816    #[mz_ore::test]
4817    fn insert_with_fills_in_place() {
4818        let pool = test_pool(usize::MAX);
4819        let want = payload(SMALL, 600);
4820        let h = pool.insert_with(SMALL, ChunkHints::default(), &TEST_CODEC, |dst| {
4821            assert_eq!(dst.len(), SMALL, "fill sees exactly the chunk length");
4822            dst.copy_from_slice(&want);
4823        });
4824        assert_eq!(h.residency(), Residency::UnbackedResident);
4825        assert_eq!(read(&h), want);
4826        pool.evict(&h);
4827        assert_eq!(read(&h), want, "round-trips through the extent");
4828
4829        // Empty and oversize take their fallback paths.
4830        let empty = pool.insert_with(0, ChunkHints::default(), &TEST_CODEC, |dst| {
4831            assert!(dst.is_empty())
4832        });
4833        assert!(read(&empty).is_empty());
4834        let big_len = (SIZE_CLASSES[SIZE_CLASSES.len() - 1] / WORD_BYTES) + 1;
4835        let big = pool.insert_with(big_len, ChunkHints::default(), &TEST_CODEC, |dst| {
4836            dst.fill(7)
4837        });
4838        assert_eq!(big.residency(), Residency::Oversize);
4839        assert_eq!(read(&big).len(), big_len);
4840    }
4841
4842    #[mz_ore::test]
4843    fn slot_exhaustion_degrades_to_heap() {
4844        // Two 64 KiB slots per class at this capacity; the third insert finds
4845        // no slot and must fall back to the heap rather than panic.
4846        let pool = Pool::with_class_capacity(128 << 10).expect("pool creation");
4847        let a = insert(&pool, &mut payload(SMALL, 700));
4848        let b = insert(&pool, &mut payload(SMALL, 701));
4849        let c = insert(&pool, &mut payload(SMALL, 702));
4850        assert_eq!(a.residency(), Residency::UnbackedResident);
4851        assert_eq!(b.residency(), Residency::UnbackedResident);
4852        assert_eq!(
4853            c.residency(),
4854            Residency::Oversize,
4855            "fallback is heap-backed"
4856        );
4857        assert_eq!(pool.stats().slot_exhausted_fallbacks, 1);
4858        assert_eq!(read(&c), payload(SMALL, 702));
4859        // Freeing a slotted chunk lets the next insert use the region again.
4860        drop(a);
4861        let d = insert(&pool, &mut payload(SMALL, 703));
4862        assert_eq!(d.residency(), Residency::UnbackedResident);
4863        assert_eq!(read(&d), payload(SMALL, 703));
4864    }
4865
4866    /// A typed body whose serialized form is its words.
4867    #[derive(Debug, PartialEq)]
4868    struct Words(Vec<u64>);
4869
4870    impl TypedBody for Words {
4871        fn words(&self) -> usize {
4872            self.0.len()
4873        }
4874
4875        fn serialize_into(&self, dst: &mut [u64]) {
4876            dst.copy_from_slice(&self.0);
4877        }
4878
4879        fn as_any(&self) -> &dyn std::any::Any {
4880            self
4881        }
4882
4883        fn into_any(self: Box<Self>) -> Box<dyn std::any::Any> {
4884            self
4885        }
4886    }
4887
4888    fn insert_typed(pool: &Pool, words: Vec<u64>) -> ChunkHandle {
4889        pool.insert_typed(Box::new(Words(words)), ChunkHints::default(), &TEST_CODEC)
4890    }
4891
4892    fn borrowed(handle: &ChunkHandle) -> Option<Vec<u64>> {
4893        handle.with_typed(|typed| {
4894            typed.map(|typed| {
4895                let words = typed.as_any().downcast_ref::<Words>().expect("test body");
4896                words.0.clone()
4897            })
4898        })
4899    }
4900
4901    fn taken(typed: Box<dyn TypedBody>) -> Vec<u64> {
4902        typed.into_any().downcast::<Words>().expect("test body").0
4903    }
4904
4905    #[mz_ore::test]
4906    fn typed_insert_borrows_and_takes_without_serializing() {
4907        let pool = test_pool(256 << 20);
4908        let orig = payload(SMALL, 800);
4909        let handle = insert_typed(&pool, orig.clone());
4910        assert_eq!(handle.residency(), Residency::UnbackedResident);
4911        assert!(handle.is_typed());
4912        assert_eq!(pool.stats().resident_bytes, 64 << 10);
4913        assert_eq!(borrowed(&handle), Some(orig.clone()));
4914        assert_eq!(read(&handle), orig, "copy-out serializes on demand");
4915        let body = handle.take_typed().expect("resident typed body");
4916        assert_eq!(taken(body), orig);
4917        let stats = pool.stats();
4918        assert_eq!(stats.resident_bytes, 0);
4919        assert_eq!(stats.live_chunks, 0);
4920        assert_eq!(stats.frees, 1);
4921        assert_eq!(stats.extent_bytes_written, 0);
4922    }
4923
4924    #[mz_ore::test]
4925    fn typed_ranged_reads_match_full_read_slice() {
4926        let pool = test_pool(256 << 20);
4927        let orig = payload(SMALL, 801);
4928        let handle = insert_typed(&pool, orig.clone());
4929        for range in [0..1, 7..SMALL - 3, SMALL - 1..SMALL, 0..SMALL] {
4930            let mut out = Vec::new();
4931            handle.read_range_into(range.clone(), &mut out);
4932            assert_eq!(out, orig[range]);
4933        }
4934    }
4935
4936    #[mz_ore::test]
4937    fn typed_evict_then_read_and_admit() {
4938        let pool = test_pool(256 << 20);
4939        let orig = payload(SMALL, 802);
4940        let handle = insert_typed(&pool, orig.clone());
4941        pool.evict(&handle);
4942        assert_eq!(handle.residency(), Residency::Evicted);
4943        assert!(!handle.is_typed());
4944        let stats = pool.stats();
4945        assert_eq!(stats.evictions_compress, 1);
4946        assert_eq!(stats.resident_bytes, 0);
4947        assert_eq!(borrowed(&handle), None);
4948        pool.poison_free_slots();
4949        assert_eq!(read(&handle), orig);
4950        assert_eq!(read_admit(&handle), orig);
4951        assert_eq!(handle.residency(), Residency::BackedResident);
4952        assert_eq!(pool.stats().admissions_budget, 1);
4953        assert_eq!(read(&handle), orig, "admitted chunk reads from its slot");
4954        let handle = handle
4955            .take_typed()
4956            .expect_err("an admitted chunk holds a slot, not a typed body");
4957        drop(handle);
4958        assert_eq!(pool.stats().resident_bytes, 0);
4959    }
4960
4961    #[mz_ore::test]
4962    fn typed_eager_backing_keeps_the_body_and_makes_eviction_cheap() {
4963        let pool = test_pool(256 << 20);
4964        let orig = payload(SMALL, 803);
4965        let handle = insert_typed(&pool, orig.clone());
4966        assert!(pool.back_step(), "one chunk is backable");
4967        assert_eq!(handle.residency(), Residency::BackedResident);
4968        assert_eq!(borrowed(&handle), Some(orig.clone()));
4969        pool.evict(&handle);
4970        assert_eq!(handle.residency(), Residency::Evicted);
4971        let stats = pool.stats();
4972        assert_eq!(stats.eager_backs, 1);
4973        assert_eq!(stats.evictions_cheap, 1);
4974        assert_eq!(stats.evictions_compress, 0);
4975        assert_eq!(stats.resident_bytes, 0);
4976        assert_eq!(read(&handle), orig);
4977    }
4978
4979    #[mz_ore::test]
4980    fn typed_take_after_backing_releases_the_extent() {
4981        let pool = test_pool(256 << 20);
4982        pool.set_rss_target(usize::MAX / 2);
4983        let orig = payload(SMALL, 804);
4984        let handle = insert_typed(&pool, orig.clone());
4985        assert!(pool.back_step());
4986        assert_eq!(handle.residency(), Residency::BackedResident);
4987        assert!(pool.stats().extent_resident_bytes > 0);
4988        let body = handle.take_typed().expect("backed typed body");
4989        assert_eq!(taken(body), orig);
4990        let stats = pool.stats();
4991        assert_eq!(stats.resident_bytes, 0);
4992        assert_eq!(stats.extent_resident_bytes, 0);
4993        assert_eq!(stats.live_chunks, 0);
4994    }
4995
4996    #[mz_ore::test]
4997    fn typed_take_races_eager_backing() {
4998        let pool = test_pool(usize::MAX);
4999        pool.set_rss_target(usize::MAX / 2);
5000        pool.set_spill_threads(4);
5001        pool.set_eager_backing(true);
5002        for round in 0..rounds(20_000, 20) {
5003            let orig = payload(SMALL, round);
5004            match insert_typed(&pool, orig.clone()).take_typed() {
5005                Ok(body) => assert_eq!(taken(body), orig),
5006                // A spill thread is backing the body.
5007                Err(handle) => assert_eq!(read(&handle), orig),
5008            }
5009        }
5010        pool.set_eager_backing(false);
5011        pool.quiesce_spill();
5012        pool.join_spill_threads();
5013        assert_eq!(pool.stats().resident_bytes, 0);
5014    }
5015
5016    #[mz_ore::test]
5017    fn typed_spill_evict_round_trip() {
5018        let pool = test_pool(usize::MAX);
5019        pool.enable_spill_without_threads();
5020        let orig = payload(SMALL, 805);
5021        let handle = insert_typed(&pool, orig.clone());
5022        pool.evict(&handle);
5023        assert_eq!(handle.residency(), Residency::WriteInFlight);
5024        assert_eq!(borrowed(&handle), Some(orig.clone()), "readable in flight");
5025        let handle = handle
5026            .take_typed()
5027            .expect_err("a spill thread owns an in-flight body");
5028        assert!(pool.spill_step());
5029        assert_eq!(handle.residency(), Residency::Evicted);
5030        let stats = pool.stats();
5031        assert_eq!(stats.evictions_compress, 1);
5032        assert_eq!(stats.resident_bytes, 0);
5033        pool.poison_free_slots();
5034        assert_eq!(read(&handle), orig);
5035    }
5036
5037    #[mz_ore::test]
5038    fn typed_spill_freed_while_queued_is_elided() {
5039        let pool = test_pool(usize::MAX);
5040        pool.enable_spill_without_threads();
5041        let handle = insert_typed(&pool, payload(SMALL, 806));
5042        pool.evict(&handle);
5043        assert_eq!(handle.residency(), Residency::WriteInFlight);
5044        drop(handle);
5045        assert!(pool.spill_step());
5046        let stats = pool.stats();
5047        assert_eq!(stats.spill_cancelled, 1);
5048        assert_eq!(stats.writes_elided, 1);
5049        assert_eq!(stats.extent_bytes_written, 0);
5050        assert_eq!(stats.resident_bytes, 0);
5051    }
5052
5053    #[mz_ore::test]
5054    fn typed_budget_is_enforced_on_insert() {
5055        let budget = 4 * (64 << 10);
5056        let pool = test_pool(budget);
5057        let handles: Vec<_> = (0..16)
5058            .map(|seed| insert_typed(&pool, payload(SMALL, 900 + seed)))
5059            .collect();
5060        let stats = pool.stats();
5061        assert!(
5062            stats.resident_bytes <= u64::cast_from(budget) + insert_slack(u64::cast_from(budget)),
5063            "resident {} over budget {budget}",
5064            stats.resident_bytes,
5065        );
5066        assert!(stats.evictions_compress > 0);
5067        for (seed, handle) in handles.iter().enumerate() {
5068            assert_eq!(read(handle), payload(SMALL, 900 + u64::cast_from(seed)));
5069        }
5070        drop(handles);
5071        let stats = pool.stats();
5072        assert_eq!(stats.resident_bytes, 0);
5073        assert_eq!(stats.live_chunks, 0);
5074    }
5075
5076    #[mz_ore::test]
5077    fn typed_body_beyond_the_classes_is_serialized_to_the_heap() {
5078        let pool = test_pool(256 << 20);
5079        let words = SIZE_CLASSES[SIZE_CLASSES.len() - 1] / WORD_BYTES + 1;
5080        let orig = payload(words, 807);
5081        let handle = insert_typed(&pool, orig.clone());
5082        assert_eq!(handle.residency(), Residency::Oversize);
5083        assert!(!handle.is_typed());
5084        assert_eq!(read(&handle), orig);
5085    }
5086}
5087
5088#[cfg(all(test, target_os = "linux"))]
5089mod file_mode_tests;