diff --git a/misc/python/materialize/mzcompose/__init__.py b/misc/python/materialize/mzcompose/__init__.py index 7ca1f74bcd923..bd6936442aa08 100644 --- a/misc/python/materialize/mzcompose/__init__.py +++ b/misc/python/materialize/mzcompose/__init__.py @@ -498,6 +498,7 @@ def get_default_system_parameters( "enable_column_paged_batcher_spill", "column_paged_batcher_budget_fraction", "column_paged_batcher_lz4", + "column_paged_batcher_swap_pageout", "enable_upsert_paged_spill", "enable_lgalloc_eager_reclamation", "lgalloc_background_interval", diff --git a/misc/python/materialize/parallel_workload/action.py b/misc/python/materialize/parallel_workload/action.py index 6ad6818be1eaa..a9af4f1d9f105 100644 --- a/misc/python/materialize/parallel_workload/action.py +++ b/misc/python/materialize/parallel_workload/action.py @@ -1618,6 +1618,9 @@ def __init__( "0.25", ] self.flags_with_values["column_paged_batcher_lz4"] = BOOLEAN_FLAG_VALUES + self.flags_with_values["column_paged_batcher_swap_pageout"] = ( + BOOLEAN_FLAG_VALUES + ) self.flags_with_values["enable_upsert_paged_spill"] = BOOLEAN_FLAG_VALUES # If you are adding a new config flag in Materialize, consider using it diff --git a/src/compute-types/src/dyncfgs.rs b/src/compute-types/src/dyncfgs.rs index a5726c1dc4b0d..59817b8397680 100644 --- a/src/compute-types/src/dyncfgs.rs +++ b/src/compute-types/src/dyncfgs.rs @@ -93,6 +93,28 @@ pub const COLUMN_PAGED_BATCHER_LZ4: Config = Config::new( `enable_column_paged_batcher_spill = true`.", ); +/// Proactively evict the column-paged batcher's lz4-compressed spill chunks +/// from RSS via `MADV_PAGEOUT` when spilling to the swap backend. Only +/// meaningful when [`COLUMN_PAGED_BATCHER_LZ4`] is `true` and the active +/// backend is swap (no scratch directory): on that path the compressed bytes +/// stay resident in the process address space and currently receive no madvise +/// at all, so the kernel reclaims them only lazily under LRU pressure. +/// `MADV_PAGEOUT` instead swaps them out eagerly at spill time, holding RSS at +/// the budget rather than letting it drift up to the pressure cliff. A later +/// page-in re-faults the pages — cheap because lz4 shrank the byte volume, +/// which is what makes eager eviction pay off on this path. +/// +/// Off by default: the eager-reclaim syscall is the one kernel interaction the +/// pager design singled out as risky, so it stays gated until proven on the +/// target workload. +pub const COLUMN_PAGED_BATCHER_SWAP_PAGEOUT: Config = Config::new( + "column_paged_batcher_swap_pageout", + false, + "Eagerly evict the column-paged batcher's lz4-compressed swap-backend spill chunks from RSS \ + via `MADV_PAGEOUT` (they otherwise receive no madvise and are reclaimed only lazily). Only \ + meaningful when `column_paged_batcher_lz4 = true` and the swap backend is active.", +); + /// Whether rendering should use `mz_join_core` rather than DD's `JoinCore::join_core`. pub const ENABLE_MZ_JOIN_CORE: Config = Config::new( "enable_mz_join_core", @@ -498,4 +520,5 @@ pub fn all_dyncfgs(configs: ConfigSet) -> ConfigSet { .add(&ENABLE_COLUMN_PAGED_BATCHER_SPILL) .add(&COLUMN_PAGED_BATCHER_BUDGET_FRACTION) .add(&COLUMN_PAGED_BATCHER_LZ4) + .add(&COLUMN_PAGED_BATCHER_SWAP_PAGEOUT) } diff --git a/src/compute/src/compute_state.rs b/src/compute/src/compute_state.rs index 0ee4cff9b272c..02da502649a0f 100644 --- a/src/compute/src/compute_state.rs +++ b/src/compute/src/compute_state.rs @@ -323,6 +323,7 @@ impl ComputeState { let enabled = ENABLE_COLUMN_PAGED_BATCHER_SPILL.get(config); let codec = COLUMN_PAGED_BATCHER_LZ4.get(config).then_some(Codec::Lz4); + let swap_pageout = COLUMN_PAGED_BATCHER_SWAP_PAGEOUT.get(config); // Budget derivation: fraction × announced memory limit, with a // 128 MiB floor so the no-pressure case doesn't page per chunk. @@ -344,12 +345,13 @@ impl ComputeState { enabled, ?backend, ?codec, + swap_pageout, fraction, mem_limit, budget_bytes = total, "column-paged batcher: applying tiered config", ); - apply_tiered_config(enabled, total, backend, codec); + apply_tiered_config(enabled, total, backend, codec, swap_pageout); } // Remember the maintenance interval locally to avoid reading it from the config set on diff --git a/src/ore/src/pager.rs b/src/ore/src/pager.rs index a2e6ae72b388b..e0755e5f8c84b 100644 --- a/src/ore/src/pager.rs +++ b/src/ore/src/pager.rs @@ -21,6 +21,7 @@ mod file; mod swap; pub use file::set_scratch_dir; +pub use swap::advise_pageout; use crate::pager::file::FileInner; use crate::pager::swap::SwapInner; diff --git a/src/ore/src/pager/swap.rs b/src/ore/src/pager/swap.rs index b3e36f4f8d58b..2a3203fa316ca 100644 --- a/src/ore/src/pager/swap.rs +++ b/src/ore/src/pager/swap.rs @@ -59,20 +59,62 @@ pub(crate) fn pageout_swap(chunks: &mut [Vec]) -> Handle { Handle::from_swap(SwapInner::new(taken)) } +/// Proactively reclaims (swaps out) the resident pages of `bytes` via +/// `MADV_PAGEOUT`, holding RSS at the caller's budget right now rather than +/// waiting for kernel LRU to reclaim under pressure the way `pageout_swap`'s +/// `MADV_COLD` hint does. +/// +/// Unlike `pageout_swap`, this takes a borrow and does **not** transfer +/// ownership: the allocation stays addressable in the caller's address space, +/// so a later read simply re-faults the swapped-out pages back in. That suits a +/// buffer the caller must keep reachable — e.g. the column pager's +/// lz4-compressed bytes kept in memory — but still wants evicted eagerly so the +/// budget is real instead of a fiction the kernel only honors at the pressure +/// cliff. +/// +/// On non-Linux targets this is a no-op (matching `MADV_COLD`). +pub fn advise_pageout(bytes: &[u8]) { + madvise_pageout(bytes); +} + #[cfg(target_os = "linux")] fn madvise_cold(chunk: &[u64]) { - if chunk.is_empty() { - return; - } - let page = page_size(); - let base_ptr = chunk.as_ptr(); - let base_addr = base_ptr.addr(); // `Vec` cannot exceed `isize::MAX` bytes, so this multiplication // cannot overflow on any supported target. Use `checked_mul` for // defense-in-depth: a corrupted length should fail loudly, not wrap. let Some(len_bytes) = chunk.len().checked_mul(std::mem::size_of::()) else { return; }; + // SAFETY: `(ptr, len_bytes)` describes the live `&[u64]` exactly. + unsafe { madvise_aligned(chunk.as_ptr().cast::(), len_bytes, libc::MADV_COLD) } +} + +#[cfg(target_os = "linux")] +fn madvise_pageout(bytes: &[u8]) { + // SAFETY: `(ptr, len)` describes the live `&[u8]` exactly. + unsafe { madvise_aligned(bytes.as_ptr(), bytes.len(), libc::MADV_PAGEOUT) } +} + +/// Issues `madvise(advice)` over the page-aligned interior of the byte range +/// `[base_ptr, base_ptr + len_bytes)`. `madvise` operates at page granularity, +/// so the start rounds up and the end rounds down to page boundaries; a range +/// that contains no whole page is skipped so we never advise pages we only +/// partially own. +/// +/// # Safety +/// +/// `base_ptr` must point to the start of a live allocation of at least +/// `len_bytes` bytes that stays valid for the duration of the call. `advice` +/// must be a non-mutating hint (`MADV_COLD`/`MADV_PAGEOUT`): both only change +/// the kernel's reclaim decision and leave the bytes readable, so concurrent +/// reads of the range remain sound. +#[cfg(target_os = "linux")] +unsafe fn madvise_aligned(base_ptr: *const u8, len_bytes: usize, advice: libc::c_int) { + if len_bytes == 0 { + return; + } + let page = page_size(); + let base_addr = base_ptr.addr(); // Round the start up and the end down to page boundaries. Both additions // use `checked_add` so that an allocation sitting near the top of the // address space can never silently wrap into a tiny range. @@ -91,23 +133,26 @@ fn madvise_cold(chunk: &[u64]) { // SAFETY: `aligned_start_addr` lies in `[base_addr, base_addr + len_bytes]` // by construction (rounding up the start cannot exceed `end_unaligned`, // which equals `base_addr + len_bytes`; the early-return above guarantees - // `start ≤ end`). That interval is exactly the range covered by the live - // `&[u64]`, so `byte_add` stays in-bounds and preserves provenance. + // `start ≤ end`). That interval is within the live allocation the caller + // promised, so `byte_add` stays in-bounds and preserves provenance. let aligned_ptr = unsafe { base_ptr.byte_add(aligned_start_addr - base_addr) } .cast::() .cast_mut(); // SAFETY: pointer/length describe a fully page-aligned subrange contained - // within the live `&[u64]` (justified above). `MADV_COLD` is non-mutating; - // it only signals reclaim preference to the kernel, so concurrent reads - // of the slice remain sound. + // within the live allocation (justified above). The caller guarantees + // `advice` is a non-mutating reclaim hint, so concurrent reads of the range + // remain sound. unsafe { - libc::madvise(aligned_ptr, aligned_len, libc::MADV_COLD); + libc::madvise(aligned_ptr, aligned_len, advice); } } #[cfg(not(target_os = "linux"))] fn madvise_cold(_chunk: &[u64]) {} +#[cfg(not(target_os = "linux"))] +fn madvise_pageout(_bytes: &[u8]) {} + #[cfg(target_os = "linux")] fn page_size() -> usize { // SAFETY: `sysconf` with a valid argument is safe. @@ -250,4 +295,25 @@ mod tests { take_swap(h, &mut dst); assert_eq!(dst, vec![1, 2, 3, 4, 5]); } + + #[mz_ore::test] + #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `madvise` on OS `linux` + fn advise_pageout_leaves_bytes_readable() { + // `MADV_PAGEOUT` is a reclaim hint: the bytes must remain addressable + // and unchanged afterwards (a read re-faults the pages back in). Use a + // multi-page buffer so the page-aligned interior is non-empty. + let pattern = |i: usize| u8::try_from(i % 251).expect("0..251 fits in u8"); + let bytes: Vec = (0..64 * 1024).map(pattern).collect(); + advise_pageout(&bytes); + // Re-read after the advice; contents are preserved. + assert!(bytes.iter().enumerate().all(|(i, &b)| b == pattern(i))); + } + + #[mz_ore::test] + fn advise_pageout_empty_and_subpage_are_noops() { + // Neither an empty slice nor a sub-page slice contains a whole page, so + // both skip the syscall entirely; they must not panic. + advise_pageout(&[]); + advise_pageout(&[1u8, 2, 3, 4]); + } } diff --git a/src/timely-util/src/column_pager.rs b/src/timely-util/src/column_pager.rs index cecb8a59fc917..6882c28f4eed8 100644 --- a/src/timely-util/src/column_pager.rs +++ b/src/timely-util/src/column_pager.rs @@ -36,6 +36,7 @@ pub mod metrics; pub mod policy; use std::io::{self, Read}; +use std::sync::atomic::{AtomicBool, Ordering}; use std::sync::{Arc, LazyLock, RwLock}; use columnar::Columnar; @@ -257,6 +258,17 @@ impl PagingPolicy for AlwaysResidentPolicy { static GLOBAL_PAGER: LazyLock> = LazyLock::new(|| RwLock::new(ColumnPager::disabled())); +/// Process-global toggle for `MADV_PAGEOUT` on the lz4 + swap spill path. +/// +/// When set, [`ColumnPager::page`] issues `MADV_PAGEOUT` over the compressed +/// bytes it keeps resident in [`CompressedInner::Memory`], proactively evicting +/// them at spill time instead of leaving them for lazy kernel reclaim. A single +/// process-global flag (set by [`apply_tiered_config`]) mirrors the backend / +/// codec selection: it is a process-wide operational choice, not a per-column +/// one, and every consumer of the shared pager reads the same value. Defaults +/// to off; the eager-reclaim syscall stays gated until proven. +static SWAP_PAGEOUT: AtomicBool = AtomicBool::new(false); + /// Install `pager` as the process-wide active pager. Subsequent /// [`global_pager`] calls return a clone of this value across all threads. /// @@ -300,12 +312,18 @@ pub fn tiered_policy() -> &'static policy::TieredPolicy { /// in-flight tickets still credit the singleton, which is harmless: the /// budget grows above the configured total until the next enable reconciles /// it via `reconfigure`. +/// +/// `swap_pageout` toggles `MADV_PAGEOUT` on the lz4 + swap spill path (see +/// `SWAP_PAGEOUT`); it is stored unconditionally so the next `page` call +/// observes it regardless of `enabled`. pub fn apply_tiered_config( enabled: bool, total_budget: usize, backend: Backend, codec: Option, + swap_pageout: bool, ) { + SWAP_PAGEOUT.store(swap_pageout, Ordering::Relaxed); let p: &Arc = &TIERED_POLICY; p.reconfigure(total_budget, backend, codec); if enabled { @@ -431,7 +449,21 @@ impl ColumnPager { codec: Some(Codec::Lz4), }); let inner = match backend { - Backend::Swap => CompressedInner::Memory(out), + Backend::Swap => { + // The compressed bytes stay resident in our own address + // space (we read them back in `take` via `FrameDecoder`), + // so the pager's ownership-transferring `pageout` does not + // fit. When `SWAP_PAGEOUT` is set, hint `MADV_PAGEOUT` + // instead: it proactively swaps the pages out now, holding + // RSS at the budget rather than leaving them as unmanaged + // anonymous memory the kernel only reclaims lazily at the + // pressure cliff. A later read re-faults them back in — + // cheap, since lz4 shrank the byte volume. + if SWAP_PAGEOUT.load(Ordering::Relaxed) { + pager::advise_pageout(&out); + } + CompressedInner::Memory(out) + } Backend::File => { // The pager deals in `Vec`, so the framed bytes // must be widened. `out` is already compressed (~4x @@ -640,6 +672,31 @@ mod tests { assert_eq!(collect_i64(&rt), (0i64..1024).collect::>()); } + /// With the swap-pageout flag on, the lz4 + swap path issues `MADV_PAGEOUT` + /// over the compressed bytes; the round-trip must still reproduce the input + /// (the advice is a non-destructive reclaim hint). Drives the global pager + /// through `apply_tiered_config` — the only path that sets the flag — with a + /// zero budget so every column spills. Resets the globals on the way out so + /// peer tests see the default disabled pager. + #[mz_ore::test] + #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `madvise` on OS `linux` + fn round_trip_swap_lz4_pageout() { + apply_tiered_config(true, 0, Backend::Swap, Some(Codec::Lz4), true); + let cp = global_pager(); + let mut col = sample_typed(); + let paged = cp.page(&mut col); + assert!(matches!( + paged, + PagedColumn::Compressed { + inner: CompressedInner::Memory(_), + .. + } + )); + let rt = cp.take(paged); + assert_eq!(collect_i64(&rt), (0i64..1024).collect::>()); + apply_tiered_config(false, 0, Backend::Swap, None, false); + } + #[mz_ore::test] #[cfg_attr(miri, ignore)] // unsupported operation: can't call foreign function `writev` on OS `linux` fn round_trip_file_uncompressed() {