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;