diff --git a/packages/bun-types/overrides.d.ts b/packages/bun-types/overrides.d.ts index ec4f620289d9..9a5608805361 100644 --- a/packages/bun-types/overrides.d.ts +++ b/packages/bun-types/overrides.d.ts @@ -100,10 +100,15 @@ declare global { * of polling. * * On macOS `level` distinguishes `"warning"` from `"critical"` based on - * the kernel's memorystatus thresholds. On Linux (PSI) and Windows the - * event is always emitted with `"critical"`. On Linux, the underlying - * PSI trigger requires `CAP_SYS_RESOURCE` on kernels before 6.6; when - * unavailable the event is never emitted. + * the kernel's memorystatus thresholds. On Linux, reclaim at one of the + * process's cgroup memory limits (the `low`, `high` and `max` counters + * of cgroup v2 `memory.events`) is emitted as `"warning"`, an OOM event + * in that cgroup and a PSI memory stall (system-wide, or of the cgroup + * when only its `memory.pressure` file is writable) as `"critical"`. + * The PSI trigger requires `CAP_SYS_RESOURCE` on kernels before 6.4, and + * the cgroup counters only exist on cgroup v2. On Windows the event is + * always emitted with `"critical"`. When no source is available the + * event is never emitted. * * This listener does not keep the event loop alive. */ diff --git a/src/js/internal-for-testing.ts b/src/js/internal-for-testing.ts index 215190f16398..fe876d122ebe 100644 --- a/src/js/internal-for-testing.ts +++ b/src/js/internal-for-testing.ts @@ -673,19 +673,25 @@ export const isMemoryPressureWatcherInstalled: () => boolean = $newCppFunction( 0, ); -// True when the installed watcher registered a real OS source (a PSI trigger -// on Linux). The watcher installs silently without one when the kernel -// refuses the trigger, so isMemoryPressureWatcherInstalled() cannot tell. -export const memoryPressureWatcherHasOsBackend: () => boolean = $newRustFunction( - "memory_pressure.rs", - "jsWatcherHasOsBackend", - 0, -); +// Empty when the OS refused every source, which isMemoryPressureWatcherInstalled() +// does not show. +export const memoryPressureArmedSources: () => ("psi" | "cgroup" | "memorystatus" | "notification")[] = + $newRustFunction("memory_pressure.rs", "jsArmedSources", 0); // The exact bytes Bun writes to /proc/pressure/memory to arm its trigger. // null where there is no PSI backend (everything except Linux). export const memoryPressurePsiTrigger: () => Buffer | null = $newRustFunction("memory_pressure.rs", "jsPsiTrigger", 0); +// Linux. The first call makes `text` the baseline of the installed "cgroup" +// source. Each later call is a memory.events notification with `text` as the +// file content, at `atMs` on the source's holdoff clock. false when no +// "cgroup" source is installed. +export const memoryPressureInjectCgroupEvents: (text: string, atMs?: number) => boolean = $newRustFunction( + "memory_pressure.rs", + "jsInjectCgroupEvents", + 2, +); + export const getEventLoopStats: () => { activeTasks: number; tasks: number; diff --git a/src/runtime/jsc_hooks.rs b/src/runtime/jsc_hooks.rs index 248e0bfe178d..852269bf692d 100644 --- a/src/runtime/jsc_hooks.rs +++ b/src/runtime/jsc_hooks.rs @@ -1659,6 +1659,10 @@ unsafe fn cancel_all_timers(vm: *mut VirtualMachine) { unsafe { crate::node::node_fs_stat_watcher::StatWatcherScheduler::shutdown_for_exit(vm); } + // The `process.on("memoryPressure")` watcher serves listeners of the + // outgoing global, which nothing removes one by one. + // SAFETY: `vm` per fn contract; the event loop is still alive. + unsafe { crate::node::memory_pressure::shutdown_for_exit(vm) }; // SAFETY: `state` is the live boxed per-thread `RuntimeState`; `vm` per fn // contract. `addr_of_mut!` does not materialize a `&mut RuntimeState`. unsafe { diff --git a/src/runtime/node/memory_pressure.rs b/src/runtime/node/memory_pressure.rs index 040fc1e0cf16..d28444e9ebd2 100644 --- a/src/runtime/node/memory_pressure.rs +++ b/src/runtime/node/memory_pressure.rs @@ -7,19 +7,21 @@ //! kernel delivers `NOTE_MEMORYSTATUS_PRESSURE_WARN` / `_CRITICAL` in //! `fflags` when `kern.memorystatus_level` crosses the warn/critical //! thresholds. -//! - Linux: PSI trigger on `/proc/pressure/memory` (or the cgroup v2 -//! `memory.pressure` file for the container's own cgroup). PSI triggers -//! signal via `POLLPRI`. Requires `CAP_SYS_RESOURCE` before kernel 6.6, -//! and PSI enabled (`CONFIG_PSI=y`). If neither path can be opened for -//! writing, the watcher silently does nothing. +//! - Linux: a PSI trigger on `/proc/pressure/memory` (or the cgroup's own +//! `memory.pressure`), emitted as `critical`. Needs `CONFIG_PSI=y`, and +//! `CAP_SYS_RESOURCE` before kernel 6.4. Plus the cgroup v2 `memory.events` +//! file of the process's own cgroup, which is readable on the read-only +//! cgroup mount a container gets and signals `POLLPRI` on every counter +//! change: `low`/`high`/`max` emit `warning`, `oom*` emit `critical`. +//! A source that cannot be set up is skipped. //! - Windows: a dedicated thread blocks on //! `CreateMemoryResourceNotification(LowMemoryResourceNotification)` and //! posts back to the JS event loop when it signals, with a 30 s holdoff //! before re-waiting (the handle is level-triggered). //! -//! All three backends post a `MemoryPressureTask` to the event loop rather -//! than calling into JS from the detector, so a listener removing itself -//! during `emit()` never races with the poll/thread that produced the event. +//! Every source posts a `MemoryPressureTask` to the event loop rather than +//! calling into JS from the detector, so a listener removing itself during +//! `emit()` never races with the poll/thread that produced the event. //! //! Armed lazily on the first listener and disarmed on the last removal via //! `onDidChangeListeners` in `BunProcess.cpp`, matching how signal handlers @@ -28,10 +30,8 @@ use bun_event_loop::ConcurrentTask::{Task, task_tag}; #[cfg(any(target_os = "linux", target_os = "android"))] use bun_jsc::ArrayBuffer; -#[cfg(not(windows))] use bun_jsc::virtual_machine::VirtualMachine; -use bun_jsc::{CallFrame, JSGlobalObject, JSValue, JsResult}; -#[cfg(not(windows))] +use bun_jsc::{CallFrame, JSGlobalObject, JSValue, JsResult, bun_string_jsc}; use core::ptr::NonNull; /// Pressure level passed to JS. Values are the `NOTE_MEMORYSTATUS_PRESSURE_*` @@ -46,7 +46,7 @@ unsafe extern "C" { } /// `run_task` target for `task_tag::MemoryPressureTask`. `lvl` is the packed -/// task payload (macOS kevent `fflags`, or `level::CRITICAL` elsewhere). +/// task payload (macOS kevent `fflags`, or one `level` constant elsewhere). pub(crate) fn emit(global: &JSGlobalObject, lvl: i32) { // macOS can deliver WARN|CRITICAL together under EV_CLEAR; pick the more severe. let lvl = if lvl & level::CRITICAL != 0 || lvl & level::WARNING == 0 { @@ -71,13 +71,15 @@ fn pressure_task(lvl: i32) -> Task { Task::init(lvl as usize as *mut MemoryPressureTask) } -#[cfg(not(windows))] -fn slot(vm: &mut VirtualMachine) -> &mut Option> { +/// `RareData.memory_pressure_watcher`: the erased `Box` of the backend's watcher. +type Slot = Option>; + +fn slot(vm: &mut VirtualMachine) -> &mut Slot { vm.rare_data().memory_pressure_watcher_slot() } // ──────────────────────────────────────────────────────────────────────────── -// POSIX backend (macOS EVFILT_MEMORYSTATUS, Linux PSI) via FilePoll +// POSIX backend (macOS EVFILT_MEMORYSTATUS, Linux PSI + cgroup) via FilePoll // ──────────────────────────────────────────────────────────────────────────── #[cfg(not(windows))] @@ -92,19 +94,29 @@ mod posix { #[cfg(any(target_os = "linux", target_os = "android", target_os = "macos"))] use bun_sys::Fd; - use super::slot; + #[cfg(any(target_os = "linux", target_os = "android"))] + pub(super) use cgroup::{CgroupEvents, Counters}; - /// Stored type-erased in `RareData.memory_pressure_watcher`. `poll` is - /// `None` when the OS backend is unavailable so `isInstalled` still - /// reflects listener presence. - struct MemoryPressureWatcher { + use super::{Slot, slot}; + + /// Stored type-erased in `RareData.memory_pressure_watcher`. It exists + /// while listeners exist, even when no source could be set up. + pub(super) struct MemoryPressureWatcher { + /// macOS: the `EVFILT_MEMORYSTATUS` poll. Linux: the PSI trigger poll. poll: Option>, + #[cfg(any(target_os = "linux", target_os = "android"))] + pub(super) cgroup: Option, } - fn take_watcher(vm: &mut VirtualMachine) -> Option> { - let raw = slot(vm).take()?; - // SAFETY: slot is populated only by `install` with a `Box`. - Some(unsafe { bun_core::heap::take(raw.as_ptr().cast::()) }) + pub(super) fn watcher_mut<'a>( + vm: &mut VirtualMachine, + ) -> Option<&'a mut MemoryPressureWatcher> { + let raw = (*slot(vm))?; + // SAFETY: slot is populated only by `install` with a `Box` + // that lives until `uninstall` takes it. Callers run on the JS thread + // and never hold the reference across a call that can reach + // `uninstall` (the sources enqueue a task instead of running JS). + Some(unsafe { &mut *raw.as_ptr().cast::() }) } fn deinit_poll(poll: &mut FilePoll) { @@ -117,35 +129,33 @@ mod posix { } } - /// Build `/sys/fs/cgroup//memory.pressure` from the cgroup v2 - /// entry in `/proc/self/cgroup`. - #[cfg(any(target_os = "linux", target_os = "android"))] - fn own_cgroup_pressure_path(buf: &mut [u8]) -> Option<&bun_core::ZStr> { - use bun_sys::O; - let fd = bun_sys::open(bun_core::zstr!("/proc/self/cgroup"), O::RDONLY, 0).ok()?; - let mut read = [0u8; 256]; - let n = bun_sys::read(fd, &mut read).unwrap_or(0); - let _ = bun_sys::close(fd); - for line in bun_core::strings::split(&read[..n], b"\n") { - let Some(rest) = line.strip_prefix(b"0::") else { - continue; - }; - let rest = core::str::from_utf8(rest.strip_prefix(b"/").unwrap_or(rest)).ok()?; - return bun_core::fmt::buf_print_z( - buf, - format_args!( - "/sys/fs/cgroup/{}{}memory.pressure", - rest, - if rest.is_empty() { "" } else { "/" } - ), - ) - .ok(); + /// Poll `fd` for `POLLPRI`, which is how every source here signals. On + /// failure the fd is closed. + #[cfg(any(target_os = "linux", target_os = "android", target_os = "macos"))] + fn register_pri_poll(global: &JSGlobalObject, fd: Fd) -> Option> { + let ctx = global.bun_vm().loop_ctx(); + let poll = FilePoll::init( + ctx, + fd, + Default::default(), + Owner::new( + poll_tag::MEMORY_PRESSURE, + NonNull::<()>::dangling().as_ptr(), + ), + ); + // SAFETY: `poll` is the fresh hive slot; `platform_event_loop` is the live uws loop. + let result = + unsafe { (*poll).register(ctx.platform_event_loop(), Flags::MemoryPressure, false) }; + if result.is_err() { + // SAFETY: fresh hive slot never handed out. + deinit_poll(unsafe { &mut *poll }); + return None; } - None + NonNull::new(poll) } - /// 150 ms of "some"-stall in any 2 s window. 2 s is the minimum - /// window for unprivileged PSI triggers (kernel 6.6+). + /// 150 ms of "some"-stall in any 2 s window. 2 s is the only window an + /// unprivileged PSI trigger may use (kernel 6.4+). /// /// The trailing NUL is part of the write: `psi_write()` in /// `kernel/sched/psi.c` replaces the last byte it receives with NUL @@ -154,98 +164,110 @@ mod posix { #[cfg(any(target_os = "linux", target_os = "android"))] pub(super) const PSI_TRIGGER: &[u8] = b"some 150000 2000000\0"; - /// Open a PSI memory file and write a trigger. Tries the system-wide - /// `/proc/pressure/memory` first, then the current cgroup's file. + /// Arm a PSI trigger on the system-wide `/proc/pressure/memory`, or + /// failing that on the current cgroup's `memory.pressure`. #[cfg(any(target_os = "linux", target_os = "android"))] - fn open_psi_fd() -> Option { + fn open_psi_fd(own_cgroup: Option<&cgroup::OwnCgroup>) -> Option { use bun_sys::O; - let mut cgroup_buf = [0u8; 320]; - let paths = [ - Some(bun_core::zstr!("/proc/pressure/memory")), - own_cgroup_pressure_path(&mut cgroup_buf), - ]; - for path in paths.into_iter().flatten() { - let Ok(fd) = bun_sys::open(path, O::RDWR | O::NONBLOCK | O::CLOEXEC, 0) else { - continue; - }; - if bun_sys::write(fd, PSI_TRIGGER).is_ok() { - return Some(fd); - } - let _ = bun_sys::close(fd); + const FLAGS: i32 = O::RDWR | O::NONBLOCK | O::CLOEXEC; + bun_sys::open(bun_core::zstr!("/proc/pressure/memory"), FLAGS, 0) + .ok() + .and_then(write_psi_trigger) + .or_else(|| { + own_cgroup? + .open(b"memory.pressure", FLAGS) + .and_then(write_psi_trigger) + }) + } + + /// Returns `fd` armed, or closes it when the kernel rejects the trigger. + #[cfg(any(target_os = "linux", target_os = "android"))] + fn write_psi_trigger(fd: Fd) -> Option { + if bun_sys::write(fd, PSI_TRIGGER).is_ok() { + return Some(fd); } + let _ = bun_sys::close(fd); None } - fn register_os_watch(global: &JSGlobalObject) -> Option> { - #[cfg(any(target_os = "linux", target_os = "android"))] - let fd = open_psi_fd()?; - #[cfg(target_os = "macos")] - let fd = Fd::from_native(0); - #[cfg(not(any(target_os = "linux", target_os = "android", target_os = "macos")))] - { - let _ = global; - return None; + #[cfg(any(target_os = "linux", target_os = "android"))] + fn new_watcher(global: &JSGlobalObject) -> MemoryPressureWatcher { + let own_cgroup = cgroup::OwnCgroup::detect(); + MemoryPressureWatcher { + poll: open_psi_fd(own_cgroup.as_ref()).and_then(|fd| register_pri_poll(global, fd)), + cgroup: own_cgroup.and_then(|cg| CgroupEvents::open(global, &cg)), } - #[cfg(any(target_os = "linux", target_os = "android", target_os = "macos"))] - { - let ctx = global.bun_vm().loop_ctx(); - let poll = FilePoll::init( - ctx, - fd, - Default::default(), - Owner::new( - poll_tag::MEMORY_PRESSURE, - NonNull::<()>::dangling().as_ptr(), - ), - ); - // SAFETY: `poll` is the fresh hive slot; `platform_event_loop` is the live uws loop. - let result = unsafe { - (*poll).register(ctx.platform_event_loop(), Flags::MemoryPressure, false) - }; - if result.is_err() { - // SAFETY: fresh hive slot never handed out. - deinit_poll(unsafe { &mut *poll }); - return None; - } - NonNull::new(poll) + } + + #[cfg(target_os = "macos")] + fn new_watcher(global: &JSGlobalObject) -> MemoryPressureWatcher { + MemoryPressureWatcher { + poll: register_pri_poll(global, Fd::from_native(0)), } } + #[cfg(not(any(target_os = "linux", target_os = "android", target_os = "macos")))] + fn new_watcher(_global: &JSGlobalObject) -> MemoryPressureWatcher { + MemoryPressureWatcher { poll: None } + } + pub(super) fn install(global: &JSGlobalObject) { let vm = global.bun_vm().as_mut(); if slot(vm).is_some() { return; } - let watcher = Box::new(MemoryPressureWatcher { - poll: register_os_watch(global), - }); + let watcher = Box::new(new_watcher(global)); *slot(global.bun_vm().as_mut()) = NonNull::new(bun_core::heap::into_raw(watcher).cast()); } - /// Whether the installed watcher holds a live OS source. `install` still - /// fills the slot when no backend could be registered, so `isInstalled` - /// alone cannot tell the two apart. - pub(super) fn has_os_backend(global: &JSGlobalObject) -> bool { - let Some(raw) = *slot(global.bun_vm().as_mut()) else { - return false; - }; - // SAFETY: slot is populated only by `install` with a `Box`, - // and nothing else borrows it while this JS host call runs. - let watcher = unsafe { raw.cast::().as_ref() }; - watcher.poll.is_some() + /// For `bun:internal-for-testing`: which sources are set up. + pub(super) fn armed_sources(global: &JSGlobalObject) -> Vec<&'static str> { + #[cfg(any(target_os = "linux", target_os = "android"))] + const POLL_SOURCE: &str = "psi"; + #[cfg(not(any(target_os = "linux", target_os = "android")))] + const POLL_SOURCE: &str = "memorystatus"; + + let (poll, cgroup) = watcher_mut(global.bun_vm().as_mut()).map_or((false, false), |w| { + (w.poll.is_some(), w.has_cgroup_source()) + }); + [(poll, POLL_SOURCE), (cgroup, "cgroup")] + .into_iter() + .filter_map(|(armed, name)| armed.then_some(name)) + .collect() } - pub(super) fn uninstall(global: &JSGlobalObject) { - let Some(watcher) = take_watcher(global.bun_vm().as_mut()) else { + impl MemoryPressureWatcher { + fn has_cgroup_source(&self) -> bool { + #[cfg(any(target_os = "linux", target_os = "android"))] + { + self.cgroup.is_some() + } + #[cfg(not(any(target_os = "linux", target_os = "android")))] + { + false + } + } + } + + pub(super) fn release(slot: &mut Slot) { + let Some(raw) = slot.take() else { return; }; + // SAFETY: slot is populated only by `install` with a `Box`. + let watcher = unsafe { bun_core::heap::take(raw.as_ptr().cast::()) }; if let Some(mut poll) = watcher.poll { - // SAFETY: `on_poll` enqueues a task instead of running user JS, - // so this is never reached from inside the dispatch and no other + // SAFETY: the sources enqueue a task instead of running user JS, + // so this is never reached from inside a dispatch and no other // `&mut FilePoll` is live. deinit_poll(unsafe { poll.as_mut() }); } + #[cfg(any(target_os = "linux", target_os = "android"))] + if let Some(mut events) = watcher.cgroup { + // SAFETY: as above. The cgroup poll is only ever reached through + // the dispatch, which has returned. + deinit_poll(unsafe { events.poll.as_mut() }); + } } /// `__bun_run_file_poll` dispatch target. `fflags` is the kqueue `fflags` @@ -253,14 +275,40 @@ mod posix { pub(crate) fn on_poll(poll: &mut FilePoll, fflags: i64) { let vm = VirtualMachine::get_mut(); - // `EPOLLERR`/`EPOLLHUP` on a PSI fd means the trigger is dead (e.g. - // the cgroup was removed). kernfs reports that permanently, so tear - // the watch down instead of emitting to avoid a level-triggered spin. #[cfg(any(target_os = "linux", target_os = "android"))] - if poll.flags.contains(Flags::Eof) || poll.flags.contains(Flags::Hup) { - drop(take_watcher(vm)); - deinit_poll(poll); - return; + { + let Some(watcher) = watcher_mut(vm) else { + deinit_poll(poll); + return; + }; + if watcher + .cgroup + .as_ref() + .is_some_and(|events| events.poll.as_ptr() == core::ptr::from_mut(poll)) + { + let fd = poll.fd; + match watcher.cgroup.as_mut().map(|events| events.read(fd)) { + Some(Ok(lvl)) => { + if let Some(lvl) = lvl { + vm.enqueue_task(super::pressure_task(lvl)); + } + } + // kernfs reports a removed file as ready forever: drop + // the source rather than spin. + Some(Err(())) | None => { + watcher.cgroup = None; + deinit_poll(poll); + } + } + return; + } + // A dead PSI trigger reports `EPOLLERR` forever: drop the source + // rather than spin. + if poll.flags.contains(Flags::Eof) || poll.flags.contains(Flags::Hup) { + watcher.poll = None; + deinit_poll(poll); + return; + } } #[cfg(not(any(target_os = "linux", target_os = "android")))] @@ -274,6 +322,203 @@ mod posix { }; vm.enqueue_task(super::pressure_task(lvl)); } + + #[cfg(any(target_os = "linux", target_os = "android"))] + mod cgroup { + use core::ptr::NonNull; + use core::time::Duration; + use std::time::Instant; + + use bun_core::strings; + use bun_io::posix_event_loop::FilePoll; + use bun_jsc::JSGlobalObject; + use bun_sys::{Fd, O}; + + use super::super::level; + + /// Per level. The kernel only coalesces `memory.events` notifications + /// to one per 10 ms, and a cgroup at `memory.max` produces one per + /// reclaim pass. 2 s is also the PSI trigger window. + const CGROUP_HOLDOFF: Duration = Duration::from_secs(2); + + /// The `0::` entry of `/proc/self/cgroup`, without its leading slash. + pub(super) struct OwnCgroup { + path: [u8; 512], + len: usize, + } + + impl OwnCgroup { + /// `None` when the process is not on a cgroup v2 hierarchy. + pub(super) fn detect() -> Option { + let fd = bun_sys::open(bun_core::zstr!("/proc/self/cgroup"), O::RDONLY, 0).ok()?; + let mut read = [0u8; 1024]; + let n = bun_sys::read(fd, &mut read).unwrap_or(0); + let _ = bun_sys::close(fd); + let rest = + strings::split(&read[..n], b"\n").find_map(|line| line.strip_prefix(b"0::"))?; + let rest = rest.strip_prefix(b"/").unwrap_or(rest); + let mut path = [0u8; 512]; + let dst = path.get_mut(..rest.len())?; + dst.copy_from_slice(rest); + Some(Self { + path, + len: rest.len(), + }) + } + + /// A container in the host's cgroup namespace sees host paths in + /// `/proc/self/cgroup`, but has its own cgroup mounted at + /// `/sys/fs/cgroup`, so a missing path falls back to the root. + pub(super) fn open(&self, file: &[u8], flags: i32) -> Option { + let mut buf = [0u8; 512 + 64]; + let own = &self.path[..self.len]; + let file = core::str::from_utf8(file).ok()?; + if !own.is_empty() { + let own = core::str::from_utf8(own).ok()?; + let path = bun_core::fmt::buf_print_z( + &mut buf, + format_args!("/sys/fs/cgroup/{own}/{file}"), + ) + .ok()?; + match bun_sys::open(path, flags, 0) { + Ok(fd) => return Some(fd), + Err(err) if err.get_errno() == bun_sys::E::ENOENT => {} + Err(_) => return None, + } + } + let path = + bun_core::fmt::buf_print_z(&mut buf, format_args!("/sys/fs/cgroup/{file}")) + .ok()?; + bun_sys::open(path, flags, 0).ok() + } + } + + /// `memory.events`, summed per level. The counters never decrease, so + /// a sum grows exactly when one of its members does. + #[derive(Clone, Copy, Default, PartialEq, Eq, Debug)] + pub(in crate::node::memory_pressure) struct Counters { + /// `low` + `high` + `max`. + reclaim: u64, + /// `oom` + `oom_kill` + `oom_group_kill`. + oom: u64, + } + + impl Counters { + pub(in crate::node::memory_pressure) fn parse(text: &[u8]) -> Counters { + let mut counters = Counters::default(); + for line in strings::split(text, b"\n") { + let Some((key, value)) = strings::split_once_char(line, b' ') else { + continue; + }; + let Some(value) = core::str::from_utf8(value) + .ok() + .and_then(|v| v.trim_ascii().parse::().ok()) + else { + continue; + }; + match key { + b"low" | b"high" | b"max" => counters.reclaim += value, + b"oom" | b"oom_kill" | b"oom_group_kill" => counters.oom += value, + _ => {} + } + } + counters + } + + fn level_of_change(self, next: Counters) -> Option { + if next.oom > self.oom { + Some(level::CRITICAL) + } else if next.reclaim > self.reclaim { + Some(level::WARNING) + } else { + None + } + } + } + + pub(in crate::node::memory_pressure) struct CgroupEvents { + pub(super) poll: NonNull, + seen: Counters, + last_warning: Option, + last_critical: Option, + /// Origin of the `bun:internal-for-testing` clock, once a test + /// has taken this source over. + injected_epoch: Option, + } + + impl CgroupEvents { + pub(super) fn open(global: &JSGlobalObject, own: &OwnCgroup) -> Option { + let fd = own.open(b"memory.events", O::RDONLY | O::NONBLOCK | O::CLOEXEC)?; + let Ok(seen) = read_counters(fd) else { + let _ = bun_sys::close(fd); + return None; + }; + let poll = super::register_pri_poll(global, fd)?; + Some(Self { + poll, + seen, + last_warning: None, + last_critical: None, + injected_epoch: None, + }) + } + + /// The read is what clears the kernfs notification, so it happens + /// on every wakeup, holdoff or not. `Err`: the file is unreadable. + pub(super) fn read(&mut self, fd: Fd) -> Result, ()> { + let counters = read_counters(fd)?; + Ok(self.observe(counters, Instant::now())) + } + + /// The level to emit for the new snapshot, if any. A critical + /// also silences warnings, so a cgroup that reclaims and OOM-kills + /// at the same time reports the worse of the two. + fn observe(&mut self, counters: Counters, now: Instant) -> Option { + let lvl = self.seen.level_of_change(counters)?; + self.seen = counters; + let quiet = |last: Option| { + last.is_some_and(|at| now.saturating_duration_since(at) < CGROUP_HOLDOFF) + }; + if lvl == level::CRITICAL { + if quiet(self.last_critical) { + return None; + } + self.last_critical = Some(now); + } else { + if quiet(self.last_warning) || quiet(self.last_critical) { + return None; + } + self.last_warning = Some(now); + } + Some(lvl) + } + + /// For `bun:internal-for-testing`. The first call on a source makes + /// `counters` its baseline and starts the clock `at_ms` is read on. + pub(in crate::node::memory_pressure) fn inject( + &mut self, + counters: Counters, + at_ms: u64, + ) -> Option { + let Some(epoch) = self.injected_epoch else { + self.injected_epoch = Some(Instant::now()); + self.seen = counters; + self.last_warning = None; + self.last_critical = None; + return None; + }; + self.observe(counters, epoch + Duration::from_millis(at_ms)) + } + } + + fn read_counters(fd: Fd) -> Result { + // Six short lines. kernfs re-arms the notification on any read, + // whatever its length. + let mut buf = [0u8; 512]; + let n = bun_sys::pread(fd, &mut buf, 0).map_err(drop)?; + Ok(Counters::parse(&buf[..n])) + } + } } // ──────────────────────────────────────────────────────────────────────────── @@ -287,7 +532,8 @@ mod windows { use bun_event_loop::ConcurrentTask::ConcurrentTask; use bun_jsc::JSGlobalObject; - use bun_jsc::virtual_machine::VirtualMachine; + + use super::{Slot, slot}; type HANDLE = *mut c_void; type BOOL = i32; @@ -329,10 +575,6 @@ mod windows { thread: Option>, } - fn slot(vm: &mut VirtualMachine) -> &mut Option> { - vm.rare_data().memory_pressure_watcher_slot() - } - fn thread_main(vm: bun_jsc::VmHandle, notify: usize, shutdown: usize) { bun_core::output::Source::configure_named_thread(bun_core::zstr!("MemoryPressure")); let handles: [HANDLE; 2] = [shutdown as HANDLE, notify as HANDLE]; @@ -398,8 +640,8 @@ mod windows { *slot(global.bun_vm().as_mut()) = NonNull::new(bun_core::heap::into_raw(watcher).cast()); } - pub(super) fn uninstall(global: &JSGlobalObject) { - let Some(raw) = slot(global.bun_vm().as_mut()).take() else { + pub(super) fn release(slot: &mut Slot) { + let Some(raw) = slot.take() else { return; }; // SAFETY: slot is populated only by `install` with a `Box`. @@ -427,10 +669,29 @@ pub(crate) extern "C" fn Bun__MemoryPressure__install(global: &JSGlobalObject) { #[unsafe(no_mangle)] pub(crate) extern "C" fn Bun__MemoryPressure__uninstall(global: &JSGlobalObject) { + release(slot(global.bun_vm().as_mut())); +} + +fn release(slot: &mut Slot) { #[cfg(not(windows))] - posix::uninstall(global); + posix::release(slot); #[cfg(windows)] - windows::uninstall(global); + windows::release(slot); +} + +/// VM teardown and the `bun test --isolate` swap: the listeners the watcher +/// served belong to the outgoing global, so nothing removes them through +/// `onDidChangeListeners`. Runs while the event loop is still alive. +/// +/// # Safety +/// `vm` is the live per-thread VM, on the JS thread. +pub(crate) unsafe fn shutdown_for_exit(vm: *mut VirtualMachine) { + // Read the raw option so a VM that never used rare data does not + // allocate it here. + // SAFETY: per fn contract. + if let Some(rare) = unsafe { &mut (*vm).rare_data }.as_deref_mut() { + release(rare.memory_pressure_watcher_slot()); + } } #[unsafe(no_mangle)] @@ -467,19 +728,51 @@ pub(crate) fn js_psi_trigger(global: &JSGlobalObject, _frame: &CallFrame) -> JsR } } -/// `memoryPressureWatcherHasOsBackend()`: whether the watcher installed by -/// the current listeners registered an OS source. On Windows `install` either -/// starts the thread or leaves the slot empty, so this equals `isInstalled`. +/// `memoryPressureArmedSources()`: `"psi"` / `"cgroup"` / `"memorystatus"` / +/// `"notification"`, whichever are set up. #[bun_jsc::host_fn] -pub(crate) fn js_watcher_has_os_backend( - global: &JSGlobalObject, - _frame: &CallFrame, -) -> JsResult { +pub(crate) fn js_armed_sources(global: &JSGlobalObject, _frame: &CallFrame) -> JsResult { #[cfg(not(windows))] - let armed = posix::has_os_backend(global); + let names: Vec<&'static str> = posix::armed_sources(global); #[cfg(windows)] - let armed = Bun__MemoryPressure__isInstalled(global); - Ok(JSValue::js_boolean(armed)) + let names: Vec<&'static str> = if Bun__MemoryPressure__isInstalled(global) { + vec!["notification"] + } else { + Vec::new() + }; + JSValue::create_array_from_iter(global, names.into_iter(), |name| { + bun_string_jsc::create_utf8_for_js(global, name.as_bytes()) + }) +} + +/// `memoryPressureInjectCgroupEvents(text, atMs)`: see `CgroupEvents::inject`. +/// `false` when no cgroup source is installed. +#[bun_jsc::host_fn] +pub(crate) fn js_inject_cgroup_events( + global: &JSGlobalObject, + frame: &CallFrame, +) -> JsResult { + #[cfg(any(target_os = "linux", target_os = "android"))] + { + let [text, at_ms] = frame.arguments_as_array::<2>(); + let text = bun_core::OwnedString::new(text.to_bun_string(global)?); + let counters = posix::Counters::parse(text.to_utf8().slice()); + let at_ms = at_ms.get_number().unwrap_or(0.0).max(0.0) as u64; + + let vm = global.bun_vm().as_mut(); + let Some(events) = posix::watcher_mut(vm).and_then(|w| w.cgroup.as_mut()) else { + return Ok(JSValue::js_boolean(false)); + }; + if let Some(lvl) = events.inject(counters, at_ms) { + vm.enqueue_task(pressure_task(lvl)); + } + Ok(JSValue::js_boolean(true)) + } + #[cfg(not(any(target_os = "linux", target_os = "android")))] + { + let _ = (global, frame); + Ok(JSValue::js_boolean(false)) + } } #[cfg(not(windows))] diff --git a/test/js/node/process/process-memory-pressure.test.ts b/test/js/node/process/process-memory-pressure.test.ts index f6d971a04703..ef15ffc8f89d 100644 --- a/test/js/node/process/process-memory-pressure.test.ts +++ b/test/js/node/process/process-memory-pressure.test.ts @@ -1,13 +1,13 @@ import { describe, expect, test } from "bun:test"; import { bunEnv, bunExe, isLinux } from "harness"; -import { closeSync, constants, openSync, readFileSync, writeSync } from "node:fs"; +import { closeSync, constants, existsSync, openSync, readFileSync, writeSync } from "node:fs"; -// process.on("memoryPressure") is a Bun extension. These tests drive the -// emit path synthetically via bun:internal-for-testing since real OS memory -// pressure cannot be induced reliably (and PSI trigger creation requires -// CAP_SYS_RESOURCE on Linux kernels before 6.6, which CI containers lack). -// The one test that arms a real PSI trigger first checks that this process -// is allowed to create one, and is skipped otherwise. +// process.on("memoryPressure") is a Bun extension. Real OS memory pressure +// cannot be induced reliably, so these tests drive the emit path through +// bun:internal-for-testing. The tests that set up a real OS source first +// check that this process may use that source (a PSI trigger needs +// CAP_SYS_RESOURCE before kernel 6.4, and most CI containers deny writes to +// /proc/pressure), and are skipped otherwise. async function run(src: string) { await using proc = Bun.spawn({ @@ -20,7 +20,28 @@ async function run(src: string) { return { stdout, stderr, exitCode }; } -function canArmPsiTriggerAt(path: string): boolean { +// The cgroup v2 file Bun would use: the one in the cgroup named by the 0:: +// line of /proc/self/cgroup, or the one at the mount root when that path is +// not visible from here (a container that shares the host's cgroup namespace). +function ownCgroupFile(name: string): string | undefined { + if (!isLinux) return undefined; + let cgroup: string; + try { + cgroup = readFileSync("/proc/self/cgroup", "utf8"); + } catch { + return undefined; + } + const entry = cgroup.split("\n").find(line => line.startsWith("0::")); + if (entry === undefined) return undefined; + const rest = entry.slice("0::".length).replace(/^\//, ""); + const own = `/sys/fs/cgroup/${rest}${rest ? "/" : ""}${name}`; + if (rest && existsSync(own)) return own; + const root = `/sys/fs/cgroup/${name}`; + return existsSync(root) ? root : undefined; +} + +function canArmPsiTriggerAt(path: string | undefined): boolean { + if (path === undefined) return false; let fd: number; try { fd = openSync(path, constants.O_RDWR | constants.O_NONBLOCK); @@ -40,20 +61,35 @@ function canArmPsiTriggerAt(path: string): boolean { } // The same two files, in the same order, that Bun tries when a listener is -// added: the system-wide PSI file, then this process's own cgroup v2 file. -function canArmPsiTrigger(): boolean { - if (!isLinux) return false; - if (canArmPsiTriggerAt("/proc/pressure/memory")) return true; - let cgroup: string; +// added: the system-wide PSI file, then this process's own cgroup file. +const canArmPsiTrigger = + isLinux && (canArmPsiTriggerAt("/proc/pressure/memory") || canArmPsiTriggerAt(ownCgroupFile("memory.pressure"))); + +// memory.events exists in every cgroup except the root, and reading it is +// enough: the notification comes from kernfs, not from a trigger we write. +const canReadCgroupEvents = (() => { + const path = ownCgroupFile("memory.events"); + if (path === undefined) return false; try { - cgroup = readFileSync("/proc/self/cgroup", "utf8"); + readFileSync(path); + return true; } catch { return false; } - const entry = cgroup.split("\n").find(line => line.startsWith("0::")); - if (entry === undefined) return false; - const rest = entry.slice("0::".length).replace(/^\//, ""); - return canArmPsiTriggerAt(`/sys/fs/cgroup/${rest}${rest ? "/" : ""}memory.pressure`); +})(); + +// Every injected counter sits this far above anything the real file can hold. +// The real file stays polled while a test injects, so a real notification +// that lands between two steps reads as lower counts and changes nothing. +const FAR_ABOVE_REAL = 1_000_000_000_000; + +function memoryEvents( + counters: Partial>, +) { + const all = { low: 0, high: 0, max: 0, oom: 0, oom_kill: 0, oom_group_kill: 0, ...counters }; + return Object.entries(all) + .map(([key, value]) => `${key} ${FAR_ABOVE_REAL + value}\n`) + .join(""); } describe.concurrent("process.on('memoryPressure')", () => { @@ -153,21 +189,93 @@ describe.concurrent("process.on('memoryPressure')", () => { expect(exitCode).toBe(0); }); - test.skipIf(!canArmPsiTrigger())("first listener arms a PSI trigger on Linux", async () => { + // Which real OS sources the first listener sets up, and that the last + // removal tears them down again. Each source is asserted only where this + // test process has just confirmed the OS allows it. + const expectedSources = [...(canArmPsiTrigger ? ["psi"] : []), ...(canReadCgroupEvents ? ["cgroup"] : [])]; + test.skipIf(expectedSources.length === 0)(`first listener arms ${expectedSources.join(" and ")}`, async () => { const { stdout, stderr, exitCode } = await run(/* js */ ` - const { memoryPressureWatcherHasOsBackend } = require("bun:internal-for-testing"); + const { memoryPressureArmedSources } = require("bun:internal-for-testing"); const h = () => {}; - const before = memoryPressureWatcherHasOsBackend(); + const before = memoryPressureArmedSources(); process.on("memoryPressure", h); - const armed = memoryPressureWatcherHasOsBackend(); + const armed = memoryPressureArmedSources(); process.off("memoryPressure", h); - const after = memoryPressureWatcherHasOsBackend(); + const after = memoryPressureArmedSources(); process.stdout.write(JSON.stringify({ before, armed, after })); `); - expect({ stdout, stderr: stderr.trim() }).toEqual({ - stdout: JSON.stringify({ before: false, armed: true, after: false }), - stderr: "", - }); + expect(stderr.trim()).toBe(""); + const { before, armed, after } = JSON.parse(stdout); + // The child may be allowed a source this process could not probe, so + // only the probed sources are required. Anything else must be the other + // Linux source, at most once. + expect({ + before, + after, + armed: expectedSources.filter(source => armed.includes(source)), + unexpected: armed.filter( + (source: string, i: number) => !["psi", "cgroup"].includes(source) || armed.indexOf(source) !== i, + ), + }).toEqual({ before: [], after: [], armed: expectedSources, unexpected: [] }); + expect(exitCode).toBe(0); + }); + + // The counters in memory.events only grow, and the kernel signals the file + // on every change. The first injection replaces the counters Bun read from + // the real file when it armed, so the real values do not matter here, and + // the clock argument drives the 2 s holdoff, so nothing here waits. + test.skipIf(!canReadCgroupEvents)("cgroup memory.events changes map to levels with a holdoff", async () => { + // Each step: [file content, holdoff clock in ms, levels the listener must see]. + // The counters accumulate from step to step, as they do in the real file. + // A null content removes and re-adds the listener, which sets up a new source. + const steps: [string | null, number | null, string[]][] = []; + let counters: Parameters[0] = {}; + const moved = (changed: typeof counters, atMs: number, emitted: string[], extraLines = "") => { + counters = { ...counters, ...changed }; + steps.push([extraLines + memoryEvents(counters), atMs, emitted]); + }; + moved({ high: 5, oom_kill: 1 }, 0, []); // the counts from before the listener existed + moved({ high: 6 }, 0, ["warning"]); // reclaim at memory.high + moved({ high: 7 }, 0, []); // another one inside the warning holdoff + moved({ oom: 1 }, 0, ["critical"]); // a warning does not hold off an OOM + moved({ oom_kill: 2 }, 0, []); // another OOM inside the critical holdoff + moved({ max: 1 }, 1000, []); // reclaim inside both holdoffs + moved({ low: 1 }, 2000, ["warning"]); // reclaim once exactly the holdoff has passed + moved({ oom_group_kill: 1 }, 2000, ["critical"]); // an OOM 2 s after the previous one + moved({ oom: 2 }, 4000, ["critical"]); // an OOM 2 s after that one + moved({ max: 2 }, 5000, []); // reclaim 3 s after the last warning, but 1 s after a critical + moved({}, 9000, [], "some_future_counter 7\n"); // a counter Bun does not classify moved + moved({ high: 8 }, 9000, ["warning"], "no_value\nsome_future_counter x\n"); // bad lines are skipped, the rest still counts + steps.push([memoryEvents({ high: 1 }), 9000, []]); // a short read: below the counts already seen + steps.push([null, 0, []]); + steps.push([memoryEvents({}), null, []]); // the new source takes its baseline from the first injection (atMs omitted) + steps.push([memoryEvents({ high: 1 }), 0, ["warning"]]); // and starts with no holdoff + const { stdout, stderr, exitCode } = await run(/* js */ ` + const { memoryPressureInjectCgroupEvents } = require("bun:internal-for-testing"); + const emitted = []; + const listener = level => emitted.push(level); + process.on("memoryPressure", listener); + const results = []; + for (const [body, atMs] of ${JSON.stringify(steps.map(([body, atMs]) => [body, atMs]))}) { + let injected = null; + if (body === null) { + // No turn of the event loop between here and the next injection, so + // the new source never polls the real file before it has its baseline. + process.off("memoryPressure", listener); + process.on("memoryPressure", listener); + } else { + injected = memoryPressureInjectCgroupEvents(body, atMs ?? undefined); + // The source queues the event on the event loop. Let it run. + await new Promise(resolve => setImmediate(resolve)); + } + results.push({ injected, emitted: emitted.splice(0) }); + } + process.stdout.write(JSON.stringify(results)); + `); + expect(stderr.trim()).toBe(""); + expect(JSON.parse(stdout)).toEqual( + steps.map(([body, , emitted]) => ({ injected: body === null ? null : true, emitted })), + ); expect(exitCode).toBe(0); });