diff --git a/docs/runtime/child-process.mdx b/docs/runtime/child-process.mdx index 9c2ea35ad4ff..e7fcf521e0ba 100644 --- a/docs/runtime/child-process.mdx +++ b/docs/runtime/child-process.mdx @@ -251,6 +251,43 @@ therefore exceed `maxBuffer` only by the single read that passed it, never by whatever the process manages to write before the kill lands. This matches Node.js. +## Using maxMemory + +`maxMemory` limits how many bytes of memory the process **and all of its descendants** may use. When the tree crosses the limit, Bun sends `killSignal` (default `SIGTERM`) to every process in it. + +```ts index.ts icon="/icons/typescript.svg" +const proc = Bun.spawn({ + cmd: ["sh", "-c", "tsc -p ."], + maxMemory: 2 * 1024 * 1024 * 1024, // 2 GiB for sh + tsc + anything tsc spawns + killSignal: "SIGKILL", +}); +await proc.exited; + +const result = Bun.spawnSync({ cmd: ["make", "-j8"], maxMemory: 4 * 1024 ** 3 }); +result.exitedDueToMaxMemory; // true if the build was killed for exceeding 4 GiB +``` + +Bun uses the kernel to enforce the limit where the kernel can do it, and samples only as a fallback: + +| Platform | How the limit is enforced | +| -------- | ------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------- | +| Windows | The child runs in a Job Object. The kernel tells Bun when the tree commits more than `maxMemory`, and Bun then terminates the job. | +| Linux | When Bun may create a memory cgroup inside its own cgroup, the child starts in one and the kernel kills the tree. The limit file is `memory.limit_in_bytes` on cgroup v1 and `memory.max` on cgroup v2. Bun never moves the child outside its own cgroup, so a systemd unit or a container still covers it. In all other cases Bun samples. | +| macOS | The kernel has no memory limit for a group of processes, so one background thread samples the tree's physical footprint. | + +The sampler sleeps for as long as a child that allocates at full speed needs to use the bytes that are left, from 1 ms to 100 ms. It also counts a descendant whose parent has exited, if Bun saw it at least one time before that. `killSignal` applies when Bun sends the signal. When the kernel does the kill (Linux cgroup), or Bun terminates a Windows job, the tree stops at once, as with `SIGKILL`. + +After the process exits, `proc.exitedDueToMaxMemory` is `true` when the limit was the cause. Use it to tell a memory kill from an ordinary failure, because on Windows and in a Linux cgroup the exit code and `signalCode` do not show it. + +To read the number yourself, call `proc.memoryUsage()`. It works with or without `maxMemory`: + +```ts index.ts icon="/icons/typescript.svg" +const proc = Bun.spawn(["bun", "run", "build"]); +const { current, peak } = proc.memoryUsage(); // bytes, for the whole process tree +``` + +`memoryUsage()` is cheap, so you can call it often. On Linux it adds up `RssAnon + RssShmem` for each process, so a page that several processes share is counted one time for each of them. Before Bun kills a tree, it confirms the total with the exact proportional numbers. `current` is `0` after the process exits. `peak` is the highest value Bun has sampled. On Windows without `maxMemory`, only the process itself is counted. + ## Inter-process communication (IPC) Bun supports a direct inter-process communication channel between two `bun` processes. To receive messages from a spawned Bun subprocess, specify an `ipc` handler. diff --git a/packages/bun-types/bun.d.ts b/packages/bun-types/bun.d.ts index 77bdcb1c0bf2..48429affb944 100644 --- a/packages/bun-types/bun.d.ts +++ b/packages/bun-types/bun.d.ts @@ -7933,6 +7933,20 @@ declare module "bun" { * @default undefined (no limit) */ maxBuffer?: number; + + /** + * The maximum number of bytes of memory the process and its descendants + * may use, combined. If the tree goes over this limit, every process in + * it is killed with `killSignal` (defaults to SIGTERM). + * + * Windows: the child runs in a Job Object, the kernel tells Bun when the tree + * goes over the limit, and Bun terminates the job. Linux: when Bun may create + * a memory cgroup inside its own, the kernel kills the tree. In all other + * cases Bun samples the tree. `killSignal` applies only when Bun sends the signal. + * + * @default undefined (no limit) + */ + maxMemory?: number; } interface SpawnSyncOptions @@ -8232,6 +8246,23 @@ declare module "bun" { * Returns `undefined` until the process has exited */ resourceUsage(): ResourceUsage | undefined; + + /** + * Memory used right now by this process and all of its descendants, in bytes. + * + * `current` is `0` after the process exits. `peak` is the highest value Bun has + * sampled; with {@link SpawnOptions.maxMemory} set, Bun samples continuously. + * + * On Windows without `maxMemory`, only the process itself is counted. + */ + memoryUsage(): { current: number; peak: number }; + + /** + * `true` when Bun or the kernel killed this process tree because it went over + * {@link SpawnOptions.maxMemory}. Use this to tell a memory kill from an + * ordinary failure. On Windows and in a Linux cgroup, `signalCode` does not show it. + */ + readonly exitedDueToMaxMemory: boolean; } /** @@ -8257,6 +8288,11 @@ declare module "bun" { signalCode?: NodeJS.Signals | number; exitedDueToTimeout?: boolean; exitedDueToMaxBuffer?: boolean; + + /** + * `true` if the process tree exceeded {@link SpawnOptions.maxMemory} and was killed for it. + */ + exitedDueToMaxMemory?: boolean; pid: number; } diff --git a/packages/bun-types/overrides.d.ts b/packages/bun-types/overrides.d.ts index 6a5dd88be9a7..bf01224dbfcf 100644 --- a/packages/bun-types/overrides.d.ts +++ b/packages/bun-types/overrides.d.ts @@ -349,6 +349,18 @@ declare module "node:fs/promises" { function exists(path: Bun.PathLike): Promise; } +declare module "node:child_process" { + interface ProcessEnvOptions { + /** + * The maximum number of bytes of memory the process and its descendants + * may use, combined. If the tree goes over this limit, it is killed. + * + * @see {@link Bun.SpawnOptions.BaseOptions.maxMemory} + */ + maxMemory?: number | undefined; + } +} + declare module "node:tls" { interface BunConnectionOptions extends Omit { /** diff --git a/src/js/internal-for-testing.ts b/src/js/internal-for-testing.ts index 203b2d3d7a20..cde80753fb49 100644 --- a/src/js/internal-for-testing.ts +++ b/src/js/internal-for-testing.ts @@ -102,6 +102,9 @@ export const subprocessInternals = { subprocess: import("bun").Subprocess, kind: "stdout" | "stderr", ) => boolean, + memoryLimitRoute: $newRustFunction("subprocess.rs", "TestingAPIs.memoryLimitRoute", 1) as ( + subprocess: import("bun").Subprocess, + ) => "job" | "cgroup" | "sampler" | undefined, }; export const iniInternals = { diff --git a/src/js/node/child_process.ts b/src/js/node/child_process.ts index 6395dda0e408..82c2684b7f0a 100644 --- a/src/js/node/child_process.ts +++ b/src/js/node/child_process.ts @@ -159,6 +159,7 @@ function spawn(file, args, options) { const windowsBatchFileError = options.windowsBatchFileError; if (windowsBatchFileError) throw windowsBatchFileError; validateTimeout(options.timeout); + validateMaxMemory(options.maxMemory); validateAbortSignal(options.signal, "options.signal"); const killSignal = sanitizeKillSignal(options.killSignal); const child = new ChildProcess(); @@ -253,6 +254,7 @@ function execFile(file, args, options?, callback?) { // Validate the timeout, if present. validateTimeout(options.timeout); + validateMaxMemory(options.maxMemory); // Validate maxBuffer, if present. validateMaxBuffer(maxBuffer); @@ -267,6 +269,7 @@ function execFile(file, args, options?, callback?) { uid: options.uid, gid: options.gid, cgroup: options.cgroup, + maxMemory: options.maxMemory, windowsHide: options.windowsHide, windowsVerbatimArguments: options.windowsVerbatimArguments, shell: options.shell, @@ -553,6 +556,7 @@ function spawnSync(file, args, options?): SpawnSyncResult { // Validate the timeout, if present. validateTimeout(options.timeout); + validateMaxMemory(options.maxMemory); // Validate maxBuffer, if present. validateMaxBuffer(maxBuffer); @@ -583,6 +587,7 @@ function spawnSync(file, args, options?): SpawnSyncResult { signalCode, exitedDueToTimeout, exitedDueToMaxBuffer, + exitedDueToMaxMemory, pid, }: Omit & { exitCode: number | null; @@ -603,6 +608,7 @@ function spawnSync(file, args, options?): SpawnSyncResult { uid: options.uid, gid: options.gid, cgroup: options.cgroup, + maxMemory: options.maxMemory, windowsVerbatimArguments: options.windowsVerbatimArguments, windowsHide: options.windowsHide, argv0: options.args[0], @@ -664,6 +670,15 @@ function spawnSync(file, args, options?): SpawnSyncResult { "ETIMEDOUT", ); } + if (exitedDueToMaxMemory && error == null) { + result.error = new SystemError( + "spawnSync " + options.file + " ENOMEM (process tree reached maxMemory limit)", + options.file, + "spawnSync " + options.file, + enomemErrorCode(), + "ENOMEM", + ); + } if (exitedDueToMaxBuffer && error == null) { result.error = new SystemError( "spawnSync " + options.file + " ENOBUFS (stdout or stderr buffer reached maxBuffer size limit)", @@ -684,6 +699,7 @@ function spawnSync(file, args, options?): SpawnSyncResult { } const etimedoutErrorCode = $newRustFunction("node_util_binding.rs", "etimedoutErrorCode", 0); const enobufsErrorCode = $newRustFunction("node_util_binding.rs", "enobufsErrorCode", 0); +const enomemErrorCode = $newRustFunction("node_util_binding.rs", "enomemErrorCode", 0); /** * Spawns a file as a shell synchronously. @@ -1473,6 +1489,8 @@ class ChildProcess extends EventEmitter { uid: options.uid, gid: options.gid, cgroup: options.cgroup, + maxMemory: options.maxMemory, + killSignal: options.maxMemory == null ? undefined : sanitizeKillSignal(options.killSignal), onExit: (handle, exitCode, signalCode, err) => { this.#handle = handle; this.pid = this.#handle.pid; @@ -1939,6 +1957,12 @@ function validateArgumentsNullCheck(args, propName) { } } +function validateMaxMemory(maxMemory) { + if (maxMemory != null && maxMemory !== Infinity && !(NumberIsInteger(maxMemory) && maxMemory >= 0)) { + throw $ERR_OUT_OF_RANGE("options.maxMemory", "a non-negative integer", maxMemory); + } +} + function validateTimeout(timeout) { if (timeout != null && !(NumberIsInteger(timeout) && timeout >= 0)) { throw $ERR_OUT_OF_RANGE("timeout", "an unsigned integer", timeout); diff --git a/src/runtime/api/BunObject.classes.ts b/src/runtime/api/BunObject.classes.ts index a7e848981255..a0317ade862d 100644 --- a/src/runtime/api/BunObject.classes.ts +++ b/src/runtime/api/BunObject.classes.ts @@ -88,6 +88,10 @@ export default [ fn: "resourceUsage", length: 0, }, + memoryUsage: { + fn: "memoryUsage", + length: 0, + }, send: { fn: "doSend", length: 1, @@ -110,6 +114,9 @@ export default [ killed: { getter: "getKilled", }, + exitedDueToMaxMemory: { + getter: "getExitedDueToMaxMemory", + }, exitCode: { getter: "getExitCode", }, diff --git a/src/runtime/api/bun/js_bun_spawn_bindings.rs b/src/runtime/api/bun/js_bun_spawn_bindings.rs index c3478628474b..1585db2a13e1 100644 --- a/src/runtime/api/bun/js_bun_spawn_bindings.rs +++ b/src/runtime/api/bun/js_bun_spawn_bindings.rs @@ -361,6 +361,7 @@ fn spawn_maybe_sync( let mut gid: Option = None; let mut kill_signal: SignalCode = SignalCode::DEFAULT; let mut max_buffer: Option = None; + let mut max_memory: Option = None; #[cfg(any(target_os = "linux", target_os = "android"))] let mut cgroup: Option = None; @@ -766,6 +767,40 @@ fn spawn_maybe_sync( } } + if let Some(val) = args.get(cx.global(), "maxMemory")? { + if val.is_number() && val.as_number().is_nan() { + return Err(cx.global().throw_range_error( + val.as_number(), + bun_fmt::OutOfRangeOptions { + field_name: b"maxMemory", + msg: b"a non-negative integer", + ..Default::default() + }, + )); + } + if !val.is_undefined_or_null() + && !(val.is_number() && val.as_number().is_infinite() && val.as_number() > 0.0) + { + let bytes = cx.global().validate_integer_range::( + val, + 0, + bun_sql_jsc::jsc::IntegerRange { + min: 0, + field_name: b"maxMemory", + ..Default::default() + }, + )?; + if bytes > 0 { + if !bun_spawn::memory_watcher::SUPPORTED { + return Err(cx.global().throw(format_args!( + "maxMemory is not supported on this platform" + ))); + } + max_memory = Some(bytes); + } + } + } + if let Some(val) = args.get(cx.global(), "maxBuffer")? { if val.is_number() && val.is_finite() { // 'Infinity' does not set maxBuffer @@ -1096,6 +1131,24 @@ fn spawn_maybe_sync( let loop_handle = EventLoopHandle::init(event_loop.cast::<()>()); + if max_memory.is_some() { + if let Err(err) = bun_spawn::memory_watcher::ensure_ready() { + return Err(cx + .global() + .throw(format_args!("Failed to start the maxMemory watcher: {err}"))); + } + } + + #[cfg(any(target_os = "linux", target_os = "android"))] + let mut memory_cgroup = match (max_memory, &cgroup) { + (Some(limit), None) => bun_spawn::memory_watcher::cgroup::Cgroup::create(limit), + _ => None, + }; + #[cfg(any(target_os = "linux", target_os = "android"))] + if let Some(created) = &memory_cgroup { + cgroup = Some(CgroupTarget::Path(ZBox::from_bytes(created.path()))); + } + #[cfg(any(target_os = "linux", target_os = "android"))] let cgroup_dir = match &cgroup { Some(target) => Some(target.open(cx.global())?), @@ -1169,9 +1222,21 @@ fn spawn_maybe_sync( // SAFETY: `argv`/`env_array` are local null-terminated C-string arrays // with argv[0] non-null; valid for this call. - let mut spawned = match unsafe { - spawn::spawn_process(&spawn_options, argv.as_ptr(), env_array.as_ptr()) - } { + let spawn_result = + unsafe { spawn::spawn_process(&spawn_options, argv.as_ptr(), env_array.as_ptr()) }; + // The cgroup was our choice, not the caller's, so a child that cannot join it runs with the sampler. + #[cfg(any(target_os = "linux", target_os = "android"))] + let spawn_result = if memory_cgroup.is_some() + && matches!(&spawn_result, Ok(sys::Result::Err(err)) if err.syscall == sys::Tag::clone3) + { + memory_cgroup = None; + spawn_options.cgroup_fd = None; + // SAFETY: same arrays as the first call. + unsafe { spawn::spawn_process(&spawn_options, argv.as_ptr(), env_array.as_ptr()) } + } else { + spawn_result + }; + let mut spawned = match spawn_result { Err(err) if err == bun_spawn::Error::Sys(bun_errno::SystemErrno::EMFILE) || err == bun_spawn::Error::Sys(bun_errno::SystemErrno::ENFILE) => @@ -1330,6 +1395,10 @@ fn spawn_maybe_sync( crate::timer::EventLoopTimerTag::SubprocessTimeout, )), exited_due_to_maxbuf: Cell::new(None), + memory_watch: JsCell::new(None), + exited_due_to_max_memory: Cell::new(false), + memory_peak: Cell::new(0), + memory_route: Cell::new(None), })); // SAFETY: subprocess_ptr is a freshly-boxed Subprocess; we hold the only reference. let subprocess = unsafe { &mut *subprocess_ptr }; @@ -1626,6 +1695,14 @@ fn spawn_maybe_sync( subprocess.update_has_pending_activity(); } + if let Some(limit) = max_memory { + subprocess.watch_memory( + limit, + #[cfg(any(target_os = "linux", target_os = "android"))] + memory_cgroup.take(), + ); + } + let mut send_exit_notification = false; if !is_sync { @@ -1974,7 +2051,11 @@ fn spawn_maybe_sync( // Once the wait is being terminated (timeout, maxBuffer, bun:test // per-test timeout), stop waiting on pipe EOF; a grandchild may // still hold the write end (Node.js SyncProcessRunner::Kill()). - if did_timeout || bun_test_fired || subprocess.exited_due_to_maxbuf.get().is_some() { + if did_timeout + || bun_test_fired + || subprocess.exited_due_to_maxbuf.get().is_some() + || subprocess.killed_for_max_memory() + { subprocess.close_readable_pipes(); } } @@ -2009,6 +2090,7 @@ fn spawn_maybe_sync( }); let exited_due_to_timeout = did_timeout; let exited_due_to_max_buffer = subprocess.exited_due_to_maxbuf.get(); + let exited_due_to_max_memory = subprocess.exited_due_to_max_memory.get(); let result_pid = JSValue::js_number_from_int32(subprocess.pid()); // SAFETY: `subprocess_ptr` was produced by `heap::into_raw(Box::new(...))` // above (spawnSync path: never handed to a JS wrapper); do what the @@ -2066,6 +2148,13 @@ fn spawn_maybe_sync( }, ); } + if max_memory.is_some() { + sync_value.put( + cx.global(), + b"exitedDueToMaxMemory", + JSValue::from(exited_due_to_max_memory), + ); + } sync_value.put(cx.global(), b"pid", result_pid); Ok(sync_value) diff --git a/src/runtime/api/bun/subprocess.rs b/src/runtime/api/bun/subprocess.rs index 4ae01e9699d1..72cd5c5403be 100644 --- a/src/runtime/api/bun/subprocess.rs +++ b/src/runtime/api/bun/subprocess.rs @@ -158,6 +158,10 @@ pub struct Subprocess<'a> { pub(crate) stdout_maxbuf: Cell>>, pub(crate) stderr_maxbuf: Cell>>, pub(crate) exited_due_to_maxbuf: Cell>, + pub(crate) memory_watch: JsCell>>, + pub(crate) exited_due_to_max_memory: Cell, + pub(crate) memory_peak: Cell, + pub(crate) memory_route: Cell>, } bun_event_loop::impl_timer_owner!(Subprocess<'_>; from_timer_ptr => event_loop_timer); @@ -362,6 +366,48 @@ impl Subprocess<'_> { this.create_resource_usage_object(global_object) } + #[bun_jsc::host_fn(method)] + pub(crate) fn memory_usage( + this: &Self, + global_object: &JSGlobalObject, + _frame: &CallFrame, + ) -> JsResult { + if !bun_spawn::memory_watcher::SUPPORTED { + return Err(global_object.throw(format_args!( + "memoryUsage() is not supported on this platform" + ))); + } + let current = if this.has_exited() { + 0 + } else if let Some(w) = this.memory_watch.get() { + let current = w.sample_now(); + this.memory_peak.set(this.memory_peak.get().max(w.peak())); + current + } else { + #[cfg(windows)] + let current = match this.process.os_handle() { + Some(handle) => bun_spawn::memory_watcher::usage(this.pid(), handle), + None => 0, + }; + #[cfg(not(windows))] + let current = bun_spawn::memory_watcher::usage(this.pid()); + this.memory_peak.set(this.memory_peak.get().max(current)); + current + }; + let object = JSValue::create_empty_object(global_object, 2); + object.put( + global_object, + b"current", + JSValue::js_number_from_uint64(current), + ); + object.put( + global_object, + b"peak", + JSValue::js_number_from_uint64(this.memory_peak.get()), + ); + Ok(object) + } + pub(crate) fn create_resource_usage_object( &self, global_object: &JSGlobalObject, @@ -657,6 +703,60 @@ impl Subprocess<'_> { crate::jsc_hooks::timer_all_mut() } + /// Call `memory_watcher::ensure_ready` before the spawn. + pub(crate) fn watch_memory( + &self, + limit: u64, + #[cfg(any(target_os = "linux", target_os = "android"))] cgroup: Option< + bun_spawn::memory_watcher::cgroup::Cgroup, + >, + ) { + if self.has_exited() { + #[cfg(any(target_os = "linux", target_os = "android"))] + if cgroup.is_some_and(|c| c.settle()) { + self.exited_due_to_max_memory.set(true); + } + return; + } + let mut opts = bun_spawn::memory_watcher::WatchOptions { + pid: self.pid(), + limit, + signal: self.kill_signal.0, + #[cfg(windows)] + process: self.process.os_handle().unwrap_or(core::ptr::null_mut()), + #[cfg(any(target_os = "linux", target_os = "android"))] + cgroup, + }; + let w = bun_spawn::memory_watcher::watch(&mut opts); + self.memory_route.set(Some(w.route())); + self.memory_watch.set(Some(w)); + } + + fn unwatch_memory(&self) { + let Some(w) = self.memory_watch.replace(None) else { + return; + }; + // A child that outlives this object, such as after a Worker ends, keeps its limit: the watcher holds its own reference. + if !self.has_exited() { + return; + } + w.unwatch(); + self.memory_peak.set(self.memory_peak.get().max(w.peak())); + if w.exceeded() { + self.exited_due_to_max_memory.set(true); + } + } + + /// True as soon as the limit was crossed, which is before the exit is reported. + pub(crate) fn killed_for_max_memory(&self) -> bool { + self.exited_due_to_max_memory.get() + || self + .memory_watch + .get() + .as_ref() + .is_some_and(|w| w.exceeded()) + } + pub(crate) fn timeout_callback(&self) { self.set_event_loop_timer_refd(false); if self.event_loop_timer.get().state == EventLoopTimerState::CANCELLED { @@ -861,6 +961,11 @@ impl Subprocess<'_> { JSValue::from(this.has_killed()) } + #[bun_jsc::host_fn(getter)] + pub(crate) fn get_exited_due_to_max_memory(this: &Self, _global: &JSGlobalObject) -> JSValue { + JSValue::from(this.killed_for_max_memory()) + } + #[bun_jsc::host_fn(getter)] pub(crate) fn get_stdio(this: &Self, global: &JSGlobalObject) -> JsResult { let array = JSValue::create_empty_array(global, 0)?; @@ -973,6 +1078,7 @@ impl Subprocess<'_> { Self::timer_all().remove(self.event_loop_timer.as_ptr()); } self.set_event_loop_timer_refd(false); + self.unwatch_memory(); // SAFETY: `jsc_vm` is the live VM owning `global_this`; mutator-thread // only. `process` is the raw `*mut Process` threaded from the vtable @@ -1075,6 +1181,7 @@ impl Subprocess<'_> { // hold the write end and the caller already opted into a bounded wait. if self.event_loop_timer.get().state == EventLoopTimerState::FIRED || self.exited_due_to_maxbuf.get().is_some() + || self.exited_due_to_max_memory.get() || self.flags.get().contains(Flags::ABORT_SIGNAL_KILLED) { self.close_readable_pipes(); @@ -1353,6 +1460,7 @@ impl Subprocess<'_> { Self::timer_all().remove(self.event_loop_timer.as_ptr()); } self.set_event_loop_timer_refd(false); + self.unwatch_memory(); let mut mb = self.stdout_maxbuf.get(); MaxBuf::MaxBuf::remove_from_subprocess(&mut mb); @@ -1569,6 +1677,26 @@ pub(crate) extern "C" fn on_pipe_close(this: *mut bun_sys::windows::libuv::Pipe) pub(crate) mod testing_apis { use super::*; + /// `"job"`, `"cgroup"` or `"sampler"` for a child spawned with `maxMemory`, so tests can require the kernel route. + #[bun_jsc::host_fn] + pub(crate) fn memory_limit_route( + global_this: &JSGlobalObject, + callframe: &CallFrame, + ) -> JsResult { + let [subprocess_value] = callframe.arguments_as_array::<1>(); + let Some(subprocess_ptr) = Subprocess::from_js(subprocess_value) else { + return Err(global_this.throw(format_args!("first argument must be a Subprocess"))); + }; + // SAFETY: `from_js` returned a live `*mut Subprocess` owned by the JS wrapper. + let subprocess = unsafe { &*subprocess_ptr }; + match subprocess.memory_route.get() { + Some(route) => { + bun_jsc::StringJsc::to_js(&bun_core::String::static_(route), global_this) + } + None => Ok(JSValue::UNDEFINED), + } + } + /// Inject a synthetic read error into a subprocess's stdout/stderr /// PipeReader, as if the underlying read() syscall (Posix) or libuv read /// callback (Windows) had failed with EBADF. Used by tests to exercise diff --git a/src/runtime/node/node_util_binding.rs b/src/runtime/node/node_util_binding.rs index 4c5f190d98d2..44c78ef94b09 100644 --- a/src/runtime/node/node_util_binding.rs +++ b/src/runtime/node/node_util_binding.rs @@ -55,6 +55,11 @@ pub(crate) fn enobufs_error_code( Ok(JSValue::js_number_from_int32(-UV_E::NOBUFS)) } +#[bun_jsc::host_fn] +pub(crate) fn enomem_error_code(_global: &JSGlobalObject, _frame: &CallFrame) -> JsResult { + Ok(JSValue::js_number_from_int32(-UV_E::NOMEM)) +} + #[bun_jsc::host_fn] pub(crate) fn uv_translate_sys_error( _global: &JSGlobalObject, diff --git a/src/spawn/lib.rs b/src/spawn/lib.rs index 8fc508cd41e7..036105e20503 100644 --- a/src/spawn/lib.rs +++ b/src/spawn/lib.rs @@ -38,6 +38,7 @@ pub mod posix_spawn { /// Ctrl+C handling for a process acting as a shell for foreground children. #[path = "ctrl_c.rs"] pub mod ctrl_c; +pub mod memory_watcher; /// `Process` / `Poller` / `WaiterThread` / `spawn_process` / `sync` / /// `Status` / `SpawnOptions` / `SpawnResult`. #[path = "process.rs"] diff --git a/src/spawn/memory_watcher.rs b/src/spawn/memory_watcher.rs new file mode 100644 index 000000000000..8d324462dc5b --- /dev/null +++ b/src/spawn/memory_watcher.rs @@ -0,0 +1,1256 @@ +//! `Bun.spawn({ maxMemory })`: kill a child tree that crosses its memory limit. The kernel enforces it where it can (Windows job, Linux cgroup); one thread samples otherwise. + +use core::sync::atomic::{AtomicBool, AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::{Duration, Instant}; + +use bun_threading::{Condvar, Guarded}; + +use crate::process::PidT; + +/// False where this module cannot measure a process tree, so `maxMemory` and `memoryUsage()` must be refused. +pub const SUPPORTED: bool = cfg!(any( + target_os = "macos", + target_os = "linux", + target_os = "android", + windows +)); + +/// The start time of a pid. `Unknown` is a read that failed for a reason other than "no such process", so it must not end a watch. +#[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))] +#[derive(Clone, Copy, PartialEq, Eq)] +enum Identity { + Is(u64), + Gone, + Unknown, +} + +/// A process seen in the tree, with its start time so a recycled pid is never counted or signalled. +#[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))] +struct Member { + pid: PidT, + identity: u64, + /// False until `kill` reached it, so a later sample can signal a process that was new or could not be read. + signalled: bool, +} + +pub struct Watch { + #[cfg(not(windows))] + pid: PidT, + limit: u64, + #[cfg(not(windows))] + signal: u8, + #[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))] + identity: u64, + /// Kept across samples so a descendant that is reparented to init stays in the tree. + #[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))] + members: Guarded>, + last: AtomicU64, + peak: AtomicU64, + exceeded: AtomicBool, + done: AtomicBool, + #[cfg(any(target_os = "linux", target_os = "android"))] + cgroup: Option, + #[cfg(windows)] + id: usize, + #[cfg(windows)] + job: bun_sys::windows::HANDLE, + #[cfg(windows)] + process: bun_sys::windows::HANDLE, + #[cfg(windows)] + notified: AtomicBool, +} + +// SAFETY: HANDLEs are process-global kernel object references; every other field is atomic, locked or immutable. +#[cfg(windows)] +unsafe impl Send for Watch {} +#[cfg(windows)] +unsafe impl Sync for Watch {} + +impl Watch { + /// What enforces the limit: `"job"` or `"cgroup"` when the kernel does, `"sampler"` when this module's thread does. + pub fn route(&self) -> &'static str { + #[cfg(windows)] + if self.notified.load(Ordering::Acquire) { + return "job"; + } + #[cfg(any(target_os = "linux", target_os = "android"))] + if self.cgroup.is_some() { + return "cgroup"; + } + "sampler" + } + + pub fn exceeded(&self) -> bool { + self.exceeded.load(Ordering::Acquire) + } + + pub fn peak(&self) -> u64 { + self.peak.load(Ordering::Acquire) + } + + /// Call when the root process has exited, before its pid can be reused. + pub fn unwatch(&self) { + self.done.store(true, Ordering::Release); + #[cfg(any(target_os = "linux", target_os = "android"))] + if let Some(cgroup) = &self.cgroup { + if cgroup.settle() { + self.exceeded.store(true, Ordering::Release); + } + } + #[cfg(windows)] + os::finish(self); + } + + /// The tree's memory right now. This runs on the JS thread, so it uses only the cheap per-process numbers. + pub fn sample_now(&self) -> u64 { + self.sample() + } + + fn sample(&self) -> u64 { + // A tick that could not be measured changes nothing: it must not kill, and it must not end the watch. + let Some(usage) = self.measure() else { + return self.last.load(Ordering::Acquire); + }; + self.last.store(usage, Ordering::Release); + self.peak.fetch_max(usage, Ordering::AcqRel); + // After the first kill, later samples signal any member that is new or was missed. + if usage > self.limit || self.exceeded() { + self.kill(); + } + usage + } + + /// False when the root is gone or its pid now names another process. `None` when that could not be read. + #[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))] + fn root_is_alive(&self) -> Option { + match os::identity(self.pid) { + Identity::Unknown => None, + Identity::Gone => Some(false), + Identity::Is(start) => Some(self.identity == 0 || start == self.identity), + } + } + + #[cfg(any(target_os = "linux", target_os = "android"))] + fn measure(&self) -> Option { + if !self.root_is_alive()? { + self.done.store(true, Ordering::Release); + return None; + } + if let Some(cgroup) = &self.cgroup { + if cgroup.exceeded() { + self.kill(); + } + return Some(cgroup.usage()); + } + let estimate = self.walk(); + // The cheap sum counts a shared page once per process, so it can only be too high. Confirm before a kill. + if estimate > self.limit && !self.exceeded() { + if let Some(exact) = self.exact() { + return Some(exact); + } + } + Some(estimate) + } + + #[cfg(target_os = "macos")] + fn measure(&self) -> Option { + if !self.root_is_alive()? { + self.done.store(true, Ordering::Release); + return None; + } + Some(self.walk()) + } + + #[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "android")))] + fn measure(&self) -> Option { + Some(os::tree_usage(self)) + } + + /// Refresh `members` and return the sum of their cheap footprints. + #[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))] + fn walk(&self) -> u64 { + let mut members = self.members.lock(); + let mut walker = os::Walker::default(); + let mut order: Vec = Vec::with_capacity(members.len() + 4); + let mut kids: Vec = Vec::new(); + let mut stack: Vec = vec![self.pid]; + // A member that could not be read this time stays a member, but it is not walked or counted. + let mut unreadable: Vec = Vec::new(); + for pass in 0..2 { + while let Some(pid) = stack.pop() { + if order.contains(&pid) { + continue; + } + order.push(pid); + kids.clear(); + walker.children(pid, &mut kids); + stack.extend_from_slice(&kids); + } + if pass == 0 { + for m in members.iter().filter(|m| !order.contains(&m.pid)) { + match os::identity(m.pid) { + Identity::Is(start) if start == m.identity => stack.push(m.pid), + Identity::Unknown => unreadable.push(Member { + pid: m.pid, + identity: m.identity, + signalled: m.signalled, + }), + _ => {} + } + } + } + } + let mut total = 0u64; + let mut next: Vec = Vec::with_capacity(order.len() + unreadable.len()); + for pid in order { + let (identity, signalled) = match members.iter().find(|m| m.pid == pid) { + Some(m) => (m.identity, m.signalled), + None => match os::identity(pid) { + Identity::Is(start) => (start, false), + _ => continue, + }, + }; + total += walker.footprint(pid); + next.push(Member { + pid, + identity, + signalled, + }); + } + next.append(&mut unreadable); + *members = next; + total + } + + /// `None` on a kernel with no `smaps_rollup`. A member that cannot be read exactly uses its own cheap number, so one failed read cannot make the whole tree look too big. + #[cfg(any(target_os = "linux", target_os = "android"))] + fn exact(&self) -> Option { + if !os::has_exact_footprint() { + return None; + } + let members = self.members.lock(); + let mut walker = os::Walker::default(); + Some( + members + .iter() + .map(|m| os::exact_footprint(m.pid, &mut walker)) + .sum(), + ) + } + + /// Safe to call on every sample: each member is signalled one time, and a member that was missed is tried again. + fn kill(&self) { + if self.done.load(Ordering::Acquire) { + return; + } + self.exceeded.store(true, Ordering::Release); + #[cfg(any(target_os = "linux", target_os = "android"))] + if let Some(cgroup) = &self.cgroup { + cgroup.kill_all(); + return; + } + self.kill_tree(); + } + + /// Children first, so a shell cannot see its child die and start a replacement before its own signal arrives. + #[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))] + fn kill_tree(&self) { + for m in self + .members + .lock() + .iter_mut() + .rev() + .filter(|m| !m.signalled) + { + if os::identity(m.pid) != Identity::Is(m.identity) { + continue; + } + // SAFETY: kill(2) has no memory-safety preconditions. + let sent = unsafe { libc::kill(m.pid, self.signal as core::ffi::c_int) } == 0; + m.signalled = + sent || std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH); + } + } + + #[cfg(not(any(target_os = "macos", target_os = "linux", target_os = "android")))] + fn kill_tree(&self) { + os::kill_tree(self); + } +} + +impl Drop for Watch { + fn drop(&mut self) { + #[cfg(windows)] + os::close(self); + } +} + +static ENTRIES: Guarded>> = Guarded::new(Vec::new()); +static WAKE: Condvar = Condvar::new(); +static THREAD_STARTED: AtomicBool = AtomicBool::new(false); + +const MIN_INTERVAL: Duration = Duration::from_millis(1); +const MAX_INTERVAL: Duration = Duration::from_millis(100); + +/// A child that only touches fresh pages reaches about this rate. The sleep is how long such a child needs to use the headroom. +const WORST_CASE_BYTES_PER_SEC: u128 = 10 << 30; + +/// Never spend more than a fifth of one core on sampling, however large the trees are. +fn interval_for(headroom_bytes: u64, sample_cost: Duration) -> Duration { + let nanos = u128::from(headroom_bytes) * 1_000_000_000 / WORST_CASE_BYTES_PER_SEC; + Duration::from_nanos(nanos.min(MAX_INTERVAL.as_nanos()) as u64) + .max(MIN_INTERVAL) + .max(sample_cost * 4) +} + +#[cfg(any(target_os = "linux", target_os = "android"))] +fn has_background_work() -> bool { + cgroup::has_pending_rmdir() +} + +#[cfg(not(any(target_os = "linux", target_os = "android")))] +fn has_background_work() -> bool { + false +} + +fn run() { + let mut snapshot: Vec> = Vec::new(); + let mut guard = ENTRIES.lock(); + loop { + // A watch can own a cgroup whose drop takes this lock, so the last reference must go away after the unlock. + let mut dead: Vec> = Vec::new(); + guard.retain(|e| { + let done = e.done.load(Ordering::Acquire); + if done { + dead.push(Arc::clone(e)); + } + !done + }); + if !dead.is_empty() { + drop(guard); + drop(dead); + guard = ENTRIES.lock(); + continue; + } + if guard.is_empty() && !has_background_work() { + WAKE.wait_guarded(&mut guard); + continue; + } + snapshot.extend(guard.iter().map(Arc::clone)); + drop(guard); + + let started = Instant::now(); + let mut min_headroom = u64::MAX; + for e in &snapshot { + if e.done.load(Ordering::Acquire) { + continue; + } + let usage = e.sample(); + // A tree that is already being killed needs a retry now and then, not the 1 ms rate. + if !e.exceeded() { + min_headroom = min_headroom.min(e.limit.saturating_sub(usage)); + } + } + snapshot.clear(); + #[cfg(any(target_os = "linux", target_os = "android"))] + cgroup::retry_pending_rmdir(); + let interval = interval_for(min_headroom, started.elapsed()); + + guard = ENTRIES.lock(); + let _ = WAKE.timed_wait_guarded(&mut guard, interval.as_nanos() as u64); + } +} + +/// Start the background thread before the child exists, so a failure is an error and never a child with no limit. +pub fn ensure_ready() -> std::io::Result<()> { + if THREAD_STARTED.swap(true, Ordering::AcqRel) { + return Ok(()); + } + let spawned = std::thread::Builder::new() + .name("MemoryWatcher".into()) + .stack_size(256 * 1024) + .spawn(run); + if let Err(e) = spawned { + THREAD_STARTED.store(false, Ordering::Release); + return Err(e); + } + Ok(()) +} + +/// Memory of an unwatched child's tree. Windows counts only the root process, because there is no Job Object to ask. +pub fn usage(pid: PidT, #[cfg(windows)] process: bun_sys::windows::HANDLE) -> u64 { + #[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))] + { + os::usage_of(pid) + } + #[cfg(windows)] + { + let _ = pid; + bun_sys::windows::GetProcessMemoryInfo(process) + .map(|c| c.PagefileUsage as u64) + .unwrap_or(0) + } + #[cfg(not(any( + target_os = "macos", + target_os = "linux", + target_os = "android", + windows + )))] + { + let _ = pid; + 0 + } +} + +pub struct WatchOptions { + pub pid: PidT, + pub limit: u64, + pub signal: u8, + #[cfg(windows)] + pub process: bun_sys::windows::HANDLE, + /// The cgroup the child was spawned into, where the kernel enforces the limit. + #[cfg(any(target_os = "linux", target_os = "android"))] + pub cgroup: Option, +} + +/// Call `ensure_ready` first. +pub fn watch(opts: &mut WatchOptions) -> Arc { + let entry = Arc::new(Watch { + #[cfg(not(windows))] + pid: opts.pid, + limit: opts.limit, + #[cfg(not(windows))] + signal: opts.signal, + #[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))] + identity: match os::identity(opts.pid) { + Identity::Is(start) => start, + _ => 0, + }, + #[cfg(any(target_os = "macos", target_os = "linux", target_os = "android"))] + members: Guarded::new(Vec::new()), + last: AtomicU64::new(0), + peak: AtomicU64::new(0), + exceeded: AtomicBool::new(false), + done: AtomicBool::new(false), + #[cfg(any(target_os = "linux", target_os = "android"))] + cgroup: opts.cgroup.take(), + #[cfg(windows)] + id: os::next_id(), + #[cfg(windows)] + job: os::create_job(opts.process), + #[cfg(windows)] + process: os::duplicate(opts.process), + #[cfg(windows)] + notified: AtomicBool::new(false), + }); + // A child can allocate a lot before the thread's first tick; sample once synchronously. + entry.sample(); + + #[cfg(windows)] + if os::arm_kernel_limit(&entry) { + return entry; + } + #[cfg(any(target_os = "linux", target_os = "android"))] + if entry.cgroup.as_ref().is_some_and(|c| c.kills_whole_group()) { + return entry; + } + + ENTRIES.lock().push(Arc::clone(&entry)); + WAKE.notify_one(); + entry +} + +/// A memory cgroup that Bun creates for one child tree, so the kernel enforces `maxMemory`. +#[cfg(any(target_os = "linux", target_os = "android"))] +pub mod cgroup { + use bun_core::ZBox; + use bun_sys::{Fd, File, O}; + use core::sync::atomic::{AtomicU32, Ordering}; + + pub struct Cgroup { + path: Vec, + v2: bool, + group_kill: bool, + /// v1 only: the kernel adds to this eventfd each time this cgroup itself runs out of memory. + oom_events: Option, + own_oom: core::sync::atomic::AtomicBool, + } + + static NEXT: AtomicU32 = AtomicU32::new(0); + static PENDING_RMDIR: bun_threading::Guarded>> = + bun_threading::Guarded::new(Vec::new()); + + fn read(dir: &[u8], name: &str) -> Option> { + File::read_from(Fd::cwd(), &[dir, b"/", name.as_bytes()].concat()).ok() + } + + fn write(dir: &[u8], name: &str, value: &[u8]) -> bun_sys::Maybe<()> { + File::openat( + Fd::cwd(), + &[dir, b"/", name.as_bytes()].concat(), + O::WRONLY, + 0, + )? + .write_all(value) + } + + fn number(bytes: &[u8]) -> u64 { + core::str::from_utf8(bytes) + .ok() + .and_then(|s| s.trim().parse().ok()) + .unwrap_or(0) + } + + /// The value after `key` in a "key value" per line file such as `memory.events`. + fn field(bytes: &[u8], key: &[u8]) -> Option { + for line in bun_core::strings::split(bytes, b"\n") { + if let Some((k, v)) = bun_core::strings::split_once_char(line, b' ') { + if k == key { + return Some(number(v)); + } + } + } + None + } + + /// With swap and no swap cap, a child over its limit is swapped out and never killed. + fn has_swap() -> bool { + read(b"/proc", "swaps").is_some_and(|b| bun_core::strings::count_char(&b, b'\n') > 1) + } + + fn rmdir(path: &[u8]) -> bun_sys::Maybe<()> { + bun_sys::rmdir(&ZBox::from_bytes(path)) + } + + pub(super) fn has_pending_rmdir() -> bool { + !PENDING_RMDIR.lock().is_empty() + } + + /// Only a cgroup that still holds processes can be removed later. A directory that is gone, or not ours, never can. + fn is_busy(err: &bun_sys::Error) -> bool { + matches!(err.get_errno(), bun_sys::E::EBUSY | bun_sys::E::ENOTEMPTY) + } + + /// A cgroup cannot be removed until its killed processes are gone, so the sampler thread retries. + pub(super) fn retry_pending_rmdir() { + PENDING_RMDIR + .lock() + .retain(|path| rmdir(path).is_err_and(|err| is_busy(&err))); + } + + /// v1 has no counter for "this cgroup's own limit could not be met", so ask for the kernel's OOM event on it. + fn register_v1_oom_events(path: &[u8]) -> Option { + // SAFETY: eventfd has no memory-safety preconditions. + let raw = unsafe { libc::eventfd(0, libc::EFD_NONBLOCK | libc::EFD_CLOEXEC) }; + if raw < 0 { + return None; + } + let events = File::from_fd(Fd::from_native(raw)); + let control = File::openat( + Fd::cwd(), + &[path, b"/memory.oom_control"].concat(), + O::RDONLY, + 0, + ) + .ok()?; + let request = format!("{} {}", raw, control.handle().native()); + write(path, "cgroup.event_control", request.as_bytes()).ok()?; + Some(events) + } + + /// Only places inside our own cgroup: a sibling would leave our systemd unit or container and escape its limits. + fn candidates(name: &str) -> Vec<(Vec, bool)> { + let mut out = Vec::new(); + let Some(own) = read(b"/proc/self", "cgroup") else { + return out; + }; + for line in bun_core::strings::split(&own, b"\n") { + let Some((controllers, path)) = bun_core::strings::split_once_char(line, b':') + .and_then(|(_, rest)| bun_core::strings::split_once_char(rest, b':')) + else { + continue; + }; + let path = path.strip_suffix(b"/").unwrap_or(path); + if bun_core::strings::split(controllers, b",").any(|c| c == b"memory") { + // v1 lets a cgroup that holds processes have child cgroups with their own limit. + let dir = [b"/sys/fs/cgroup/memory", path, b"/", name.as_bytes()].concat(); + out.push((dir, false)); + } else if controllers.is_empty() && path.is_empty() { + // v2 allows that only in the real root cgroup. + out.push((format!("/sys/fs/cgroup/{name}").into_bytes(), true)); + } + } + out + } + + impl Cgroup { + /// `None` when this process may not create a memory cgroup inside its own; the caller then samples. + pub fn create(limit: u64) -> Option { + // SAFETY: getpid has no preconditions. + let pid = unsafe { libc::getpid() }; + let name = format!("bun-{pid}-{}", NEXT.fetch_add(1, Ordering::Relaxed)); + let limit = limit.to_string(); + for (path, v2) in candidates(&name) { + if bun_sys::mkdir(&ZBox::from_bytes(&path), 0o755).is_err() { + continue; + } + let limited = if v2 { + write(&path, "memory.max", limit.as_bytes()) + } else { + write(&path, "memory.limit_in_bytes", limit.as_bytes()) + }; + let swap_capped = if v2 { + write(&path, "memory.swap.max", b"0") + } else { + write(&path, "memory.memsw.limit_in_bytes", limit.as_bytes()) + }; + // A v1 cgroup must report its OOM kills (Linux 4.13) and must not inherit a disabled OOM killer, which freezes the tree at the limit. + let reports_kills = v2 + || read(&path, "memory.oom_control").is_some_and(|b| { + field(&b, b"oom_kill").is_some() + && field(&b, b"oom_kill_disable").unwrap_or(0) == 0 + }); + if limited.is_err() || !reports_kills || (swap_capped.is_err() && has_swap()) { + let _ = rmdir(&path); + continue; + } + let oom_events = if v2 { + None + } else { + register_v1_oom_events(&path) + }; + if !v2 && oom_events.is_none() { + let _ = rmdir(&path); + continue; + } + let group_kill = v2 && write(&path, "memory.oom.group", b"1").is_ok(); + return Some(Cgroup { + path, + v2, + group_kill, + oom_events, + own_oom: core::sync::atomic::AtomicBool::new(false), + }); + } + None + } + + pub fn path(&self) -> &[u8] { + &self.path + } + + /// False when the kernel kills one process at the limit and leaves the rest; the sampler thread then finishes the job. + pub(super) fn kills_whole_group(&self) -> bool { + self.group_kill + } + + /// Anonymous and shared memory, which a process cannot give back. Page cache from file I/O is left out. + pub(super) fn usage(&self) -> u64 { + let Some(stat) = read(&self.path, "memory.stat") else { + return 0; + }; + let sum = |anon: &[u8], shmem: &[u8]| { + Some(field(&stat, anon)? + field(&stat, shmem).unwrap_or(0)) + }; + if self.v2 { + sum(b"anon", b"shmem").unwrap_or(0) + } else { + sum(b"total_rss", b"total_shmem") + .or_else(|| sum(b"rss", b"shmem")) + .unwrap_or(0) + } + } + + /// A member was OOM-killed because this cgroup's own limit could not be met. A kill by the host or by a parent cgroup does not count, and a limit hit that reclaim solved does not count. + pub(super) fn exceeded(&self) -> bool { + if self.v2 { + let Some(events) = read(&self.path, "memory.events.local") + .or_else(|| read(&self.path, "memory.events")) + else { + return false; + }; + // `oom` rises only when this cgroup's own limit could not be met. `oom_group_kill` also rises for a host or parent OOM. + return field(&events, b"oom").unwrap_or(0) > 0 + && field(&events, b"oom_kill").unwrap_or(0) > 0; + } + if let Some(events) = &self.oom_events { + let mut count = [0u8; 8]; + let fired = + events.read(&mut count).is_ok_and(|n| n == 8) && u64::from_ne_bytes(count) > 0; + // The event also fires when a parent cgroup runs out of memory, so this cgroup must have reached its own limit too. + if fired && self.reached_own_limit() { + self.own_oom.store(true, Ordering::Release); + } + } + self.own_oom.load(Ordering::Acquire) + && read(&self.path, "memory.oom_control") + .is_some_and(|b| field(&b, b"oom_kill").unwrap_or(0) > 0) + } + + fn reached_own_limit(&self) -> bool { + let peak = read(&self.path, "memory.max_usage_in_bytes").map(|b| number(&b)); + let limit = read(&self.path, "memory.limit_in_bytes").map(|b| number(&b)); + matches!((peak, limit), (Some(peak), Some(limit)) if limit > 0 && peak >= limit) + } + + pub(super) fn kill_all(&self) { + if self.v2 && write(&self.path, "cgroup.kill", b"1").is_ok() { + return; + } + if let Some(procs) = read(&self.path, "cgroup.procs") { + for pid in bun_core::strings::split(&procs, b"\n") { + let pid = number(pid) as libc::pid_t; + if pid > 0 { + // SAFETY: kill(2) has no memory-safety preconditions. + unsafe { libc::kill(pid, libc::SIGKILL) }; + } + } + } + } + + /// When the kernel killed for the limit, kill what it left alive and return true. + pub fn settle(&self) -> bool { + let exceeded = self.exceeded(); + if exceeded { + self.kill_all(); + } + exceeded + } + } + + impl Drop for Cgroup { + fn drop(&mut self) { + if rmdir(&self.path).is_err_and(|err| is_busy(&err)) { + // The sampler checks this list under `ENTRIES` before it sleeps, so the push must be under it too. + let entries = super::ENTRIES.lock(); + PENDING_RMDIR.lock().push(core::mem::take(&mut self.path)); + drop(entries); + super::WAKE.notify_one(); + } + } + } +} + +#[cfg(target_os = "macos")] +mod os { + use super::{Identity, PidT}; + use core::ffi::{c_int, c_void}; + + #[repr(C)] + #[derive(Default)] + struct RusageInfoV0 { + ri_uuid: [u8; 16], + ri_user_time: u64, + ri_system_time: u64, + ri_pkg_idle_wkups: u64, + ri_interrupt_wkups: u64, + ri_pageins: u64, + ri_wired_size: u64, + ri_resident_size: u64, + ri_phys_footprint: u64, + ri_proc_start_abstime: u64, + ri_proc_exit_abstime: u64, + } + + unsafe extern "C" { + fn proc_pid_rusage(pid: c_int, flavor: c_int, buffer: *mut RusageInfoV0) -> c_int; + } + + pub(super) struct Walker { + buf: Vec, + } + + impl Default for Walker { + fn default() -> Self { + Walker { buf: vec![0; 64] } + } + } + + impl Walker { + pub(super) fn children(&mut self, pid: PidT, out: &mut Vec) { + loop { + let cap = c_int::try_from(self.buf.len() * core::mem::size_of::()) + .unwrap_or(c_int::MAX); + // SAFETY: `buf` is a writable buffer of `cap` bytes. + let n = unsafe { + bun_sys::c::proc_listchildpids(pid, self.buf.as_mut_ptr().cast::(), cap) + }; + if n <= 0 { + return; + } + let n = n as usize; + if n >= self.buf.len() { + self.buf.resize(self.buf.len() * 2, 0); + continue; + } + out.extend_from_slice(&self.buf[..n]); + return; + } + } + + /// `phys_footprint` is what Activity Monitor shows and what the kernel's own limits use. + pub(super) fn footprint(&mut self, pid: PidT) -> u64 { + let mut info = RusageInfoV0::default(); + // SAFETY: `info` is a valid out-buffer of the size `RUSAGE_INFO_V0` (flavor 0) writes. + if unsafe { proc_pid_rusage(pid, 0, &raw mut info) } != 0 { + return 0; + } + info.ri_phys_footprint + } + } + + pub(super) fn usage_of(root: PidT) -> u64 { + let mut walker = Walker::default(); + let mut stack = vec![root]; + let mut kids = Vec::new(); + let mut total = 0u64; + while let Some(pid) = stack.pop() { + total += walker.footprint(pid); + kids.clear(); + walker.children(pid, &mut kids); + stack.extend_from_slice(&kids); + } + total + } + + pub(super) fn identity(pid: PidT) -> Identity { + let mut info: bun_sys::c::struct_proc_bsdinfo = bun_core::ffi::zeroed(); + let size = core::mem::size_of::() as c_int; + // SAFETY: `info` is a valid out-buffer of `size` bytes. + let n = unsafe { + bun_sys::c::proc_pidinfo( + pid, + bun_sys::c::PROC_PIDTBSDINFO, + 0, + (&raw mut info).cast::(), + size, + ) + }; + if n == size { + return Identity::Is( + info.pbi_start_tvsec + .wrapping_mul(1_000_000) + .wrapping_add(info.pbi_start_tvusec), + ); + } + if std::io::Error::last_os_error().raw_os_error() == Some(libc::ESRCH) { + Identity::Gone + } else { + Identity::Unknown + } + } +} + +#[cfg(any(target_os = "linux", target_os = "android"))] +mod os { + use super::{Identity, PidT}; + use bun_sys::{Fd, FdExt, File, O, SizeHint}; + use core::sync::atomic::{AtomicU8, Ordering}; + + fn read_small(path: &str, buf: &mut Vec) -> bun_sys::Maybe<()> { + buf.clear(); + File::openat(Fd::cwd(), path.as_bytes(), O::RDONLY, 0)? + .read_to_end_with_array_list(buf, SizeHint::ProbablySmall) + .map(|_| ()) + } + + /// Only these errors mean that the process does not exist. A full fd table or a hidden `/proc` entry does not. + fn is_gone(err: &bun_sys::Error) -> bool { + matches!(err.get_errno(), bun_sys::E::ENOENT | bun_sys::E::ESRCH) + } + + fn parse(bytes: &[u8]) -> Option { + core::str::from_utf8(bytes).ok()?.trim().parse().ok() + } + + fn page_size() -> u64 { + // SAFETY: sysconf has no preconditions. + let n = unsafe { libc::sysconf(libc::_SC_PAGESIZE) }; + if n > 0 { n as u64 } else { 4096 } + } + + /// The fields of `/proc//stat` after the command name, which can itself hold spaces and parentheses. + fn stat_fields(buf: &[u8]) -> Option> { + let close = bun_core::strings::last_index_of_char(buf, b')')?; + Some(bun_core::strings::tokenize(&buf[close + 1..], b" ")) + } + + const UNKNOWN: u8 = 0; + const YES: u8 = 1; + const NO: u8 = 2; + /// `/proc//task//children` needs `CONFIG_PROC_CHILDREN`, which some kernels and sandboxes lack. + static HAS_CHILDREN_FILES: AtomicU8 = AtomicU8::new(UNKNOWN); + + fn has_children_files() -> bool { + match HAS_CHILDREN_FILES.load(Ordering::Relaxed) { + YES => true, + NO => false, + _ => { + // SAFETY: getpid has no preconditions. + let pid = unsafe { libc::getpid() }; + let mut buf = Vec::new(); + let yes = read_small(&format!("/proc/{pid}/task/{pid}/children"), &mut buf).is_ok(); + HAS_CHILDREN_FILES.store(if yes { YES } else { NO }, Ordering::Relaxed); + yes + } + } + } + + #[derive(Default)] + pub(super) struct Walker { + buf: Vec, + /// `(pid, parent)` for every process, built once per sample when the children files are missing. + parents: Option>, + } + + impl Walker { + pub(super) fn children(&mut self, pid: PidT, out: &mut Vec) { + if has_children_files() { + self.children_from_files(pid, out); + } else { + self.children_from_scan(pid, out); + } + } + + fn children_from_files(&mut self, pid: PidT, out: &mut Vec) { + let Ok(task_dir) = bun_sys::open_dir_absolute(format!("/proc/{pid}/task").as_bytes()) + else { + return; + }; + let mut tasks = bun_sys::iterate_dir(task_dir); + while let Ok(Some(task)) = tasks.next() { + let Ok(tid) = core::str::from_utf8(task.name.slice_u8()) else { + continue; + }; + if read_small(&format!("/proc/{pid}/task/{tid}/children"), &mut self.buf).is_err() { + continue; + } + out.extend(bun_core::strings::tokenize(&self.buf, b" ").filter_map(parse::)); + } + task_dir.close(); + } + + fn children_from_scan(&mut self, pid: PidT, out: &mut Vec) { + if self.parents.is_none() { + let mut parents = Vec::new(); + if let Ok(proc_dir) = bun_sys::open_dir_absolute(b"/proc") { + let mut entries = bun_sys::iterate_dir(proc_dir); + while let Ok(Some(entry)) = entries.next() { + let Some(child) = parse::(entry.name.slice_u8()) else { + continue; + }; + if read_small(&format!("/proc/{child}/stat"), &mut self.buf).is_err() { + continue; + } + if let Some(parent) = stat_fields(&self.buf) + .and_then(|mut f| f.nth(1)) + .and_then(parse::) + { + parents.push((child, parent)); + } + } + proc_dir.close(); + } + self.parents = Some(parents); + } + if let Some(parents) = &self.parents { + out.extend(parents.iter().filter(|(_, p)| *p == pid).map(|(c, _)| *c)); + } + } + + /// `RssAnon + RssShmem`: memory the process cannot give back. A shared page counts once per process, so a sum can be too high. + pub(super) fn footprint(&mut self, pid: PidT) -> u64 { + if read_small(&format!("/proc/{pid}/status"), &mut self.buf).is_ok() { + if let Some(kb) = sum_kb(&self.buf, &[b"RssAnon", b"RssShmem"]) { + return kb * 1024; + } + } + // Before Linux 4.5 there is no split, so use resident minus shared from statm. + if read_small(&format!("/proc/{pid}/statm"), &mut self.buf).is_err() { + return 0; + } + let mut it = bun_core::strings::tokenize(&self.buf, b" ") + .skip(1) + .map(|f| parse::(f).unwrap_or(0)); + let resident = it.next().unwrap_or(0); + let shared = it.next().unwrap_or(0); + resident.saturating_sub(shared) * page_size() + } + } + + /// The digits after any spaces and tabs, as in "\t 1234 kB". `status` uses a tab and `smaps_rollup` uses spaces. + fn leading_number(bytes: &[u8]) -> Option { + let start = bytes.iter().position(|b| !matches!(b, b' ' | b'\t'))?; + let digits = &bytes[start..]; + let end = digits + .iter() + .position(|b| !b.is_ascii_digit()) + .unwrap_or(digits.len()); + core::str::from_utf8(&digits[..end]).ok()?.parse().ok() + } + + /// The sum of the named "Key: 123 kB" lines, or `None` when the text has none of them. + fn sum_kb(text: &[u8], keys: &[&[u8]]) -> Option { + let mut total = None; + for line in bun_core::strings::split(text, b"\n") { + let Some((key, rest)) = bun_core::strings::split_once_char(line, b':') else { + continue; + }; + if keys.contains(&key) { + total = Some(total.unwrap_or(0) + leading_number(rest).unwrap_or(0)); + } + } + total + } + + static HAS_SMAPS_ROLLUP: AtomicU8 = AtomicU8::new(UNKNOWN); + + /// `smaps_rollup` exists from Linux 4.14. Without it the cheap sum is the only number there is. + pub(super) fn has_exact_footprint() -> bool { + match HAS_SMAPS_ROLLUP.load(Ordering::Relaxed) { + YES => true, + NO => false, + _ => { + let mut buf = Vec::new(); + let yes = read_small("/proc/self/smaps_rollup", &mut buf).is_ok(); + HAS_SMAPS_ROLLUP.store(if yes { YES } else { NO }, Ordering::Relaxed); + yes + } + } + } + + /// Proportional memory: each shared page is divided between the processes that map it. + pub(super) fn exact_footprint(pid: PidT, walker: &mut Walker) -> u64 { + let mut buf = Vec::with_capacity(1024); + match read_small(&format!("/proc/{pid}/smaps_rollup"), &mut buf) { + // Linux 4.14 to 5.4 has only the total `Pss`, which counts file pages, so use the cheap number there. + Ok(()) => match sum_kb(&buf, &[b"Pss_Anon", b"Pss_Shmem"]) { + Some(kb) => kb * 1024, + None => walker.footprint(pid), + }, + Err(err) if is_gone(&err) => 0, + Err(_) => walker.footprint(pid), + } + } + + pub(super) fn usage_of(root: PidT) -> u64 { + let mut walker = Walker::default(); + let mut stack = vec![root]; + let mut kids = Vec::new(); + let mut total = 0u64; + while let Some(pid) = stack.pop() { + total += walker.footprint(pid); + kids.clear(); + walker.children(pid, &mut kids); + stack.extend_from_slice(&kids); + } + total + } + + pub(super) fn identity(pid: PidT) -> Identity { + let mut buf: Vec = Vec::with_capacity(256); + match read_small(&format!("/proc/{pid}/stat"), &mut buf) { + Err(err) if is_gone(&err) => Identity::Gone, + Err(_) => Identity::Unknown, + Ok(()) => stat_fields(&buf) + .and_then(|mut f| f.nth(19)) + .and_then(parse::) + .map_or(Identity::Unknown, Identity::Is), + } + } +} + +#[cfg(windows)] +mod os { + use super::Watch; + use bun_sys::windows::{self, HANDLE}; + use core::sync::atomic::{AtomicUsize, Ordering}; + use std::sync::Arc; + + static NEXT_ID: AtomicUsize = AtomicUsize::new(1); + + /// The completion key. It is never reused, so a late message cannot reach a newer watch. + pub(super) fn next_id() -> usize { + NEXT_ID.fetch_add(1, Ordering::Relaxed) + } + + /// libuv closes its own process handle when the child is reaped, so the watch keeps its own. + pub(super) fn duplicate(process: HANDLE) -> HANDLE { + let mut out: HANDLE = core::ptr::null_mut(); + let me = windows::GetCurrentProcess(); + // SAFETY: `out` is a valid out-pointer; bad handles fail with 0. + let ok = unsafe { + windows::kernel32::DuplicateHandle( + me, + process, + me, + &raw mut out, + 0, + 0, + windows::DUPLICATE_SAME_ACCESS, + ) + }; + if ok == 0 { core::ptr::null_mut() } else { out } + } + + pub(super) fn create_job(process: HANDLE) -> HANDLE { + // SAFETY: null attributes/name are documented-valid. + let job = unsafe { windows::CreateJobObjectA(core::ptr::null_mut(), core::ptr::null()) }; + if job.is_null() { + return job; + } + // SAFETY: both handles are live. + if unsafe { windows::AssignProcessToJobObject(job, process) } == 0 { + // SAFETY: `job` was just created and is owned here. + unsafe { windows::CloseHandle(job) }; + return core::ptr::null_mut(); + } + job + } + + pub(super) fn close(w: &Watch) { + for handle in [w.job, w.process] { + if !handle.is_null() { + // SAFETY: both handles are owned by this Watch. + unsafe { windows::CloseHandle(handle) }; + } + } + } + + pub(super) fn tree_usage(w: &Watch) -> u64 { + if !w.job.is_null() { + let (current, peak) = windows::job_memory_usage(w.job); + w.peak.fetch_max(peak, Ordering::AcqRel); + if let Some(current) = current { + return current; + } + } + if w.process.is_null() { + return 0; + } + // The root has exited, so the sampler must let go of this watch. + if windows::kernel32::WaitForSingleObject(w.process, 0) == 0 { + w.done.store(true, Ordering::Release); + } + windows::GetProcessMemoryInfo(w.process) + .map(|c| c.PagefileUsage as u64) + .unwrap_or(0) + } + + static PORT: std::sync::OnceLock = std::sync::OnceLock::new(); + static NOTIFIED: bun_threading::Guarded>> = + bun_threading::Guarded::new(Vec::new()); + + fn port() -> Option { + let port = *PORT.get_or_init(|| { + let Ok(port) = windows::CreateIoCompletionPort( + windows::INVALID_HANDLE_VALUE, + core::ptr::null_mut(), + 0, + 1, + ) else { + return 0; + }; + let raw = port as usize; + let spawned = std::thread::Builder::new() + .name("MemoryWatcher".into()) + .stack_size(256 * 1024) + .spawn(move || wait_for_limits(raw as HANDLE)); + if spawned.is_err() { + // SAFETY: `port` was just created and nothing else holds it. + unsafe { windows::CloseHandle(port) }; + return 0; + } + raw + }); + (port != 0).then_some(port as HANDLE) + } + + /// The kernel posts one message when a job crosses its limit, so this thread never polls. + fn wait_for_limits(port: HANDLE) { + loop { + let mut message: windows::DWORD = 0; + let mut key: windows::ULONG_PTR = 0; + let mut overlapped: *mut windows::OVERLAPPED = core::ptr::null_mut(); + // SAFETY: all three out-pointers are valid for the call. + let ok = unsafe { + windows::kernel32::GetQueuedCompletionStatus( + port, + &raw mut message, + &raw mut key, + &raw mut overlapped, + windows::INFINITE, + ) + }; + if ok == 0 { + continue; + } + let Some(watch) = NOTIFIED.lock().iter().find(|w| w.id == key).map(Arc::clone) else { + continue; + }; + match message { + windows::JOB_OBJECT_MSG_NOTIFICATION_LIMIT => watch.kill(), + // The tree is gone. This releases a watch whose Subprocess object went away before the child did. + windows::JOB_OBJECT_MSG_ACTIVE_PROCESS_ZERO => { + watch.done.store(true, Ordering::Release); + forget(&watch); + } + _ => {} + } + } + } + + /// True when the kernel now reports the limit for this watch, so the sampler is not needed. + pub(super) fn arm_kernel_limit(entry: &Arc) -> bool { + if entry.job.is_null() { + return false; + } + let Some(port) = port() else { return false }; + NOTIFIED.lock().push(Arc::clone(entry)); + if windows::job_notify_memory_limit(entry.job, port, entry.id, entry.limit) { + entry.notified.store(true, Ordering::Release); + return true; + } + forget(entry); + false + } + + fn forget(w: &Watch) { + NOTIFIED.lock().retain(|other| other.id != w.id); + } + + /// The root exited: read the job's real peak, and stop holding the watch for a message that may never come. + pub(super) fn finish(w: &Watch) { + let _ = tree_usage(w); + forget(w); + } + + pub(super) fn kill_tree(w: &Watch) { + // SAFETY: `job` is a live job handle. + if !w.job.is_null() && unsafe { windows::TerminateJobObject(w.job, 1) } != 0 { + return; + } + if !w.process.is_null() { + // SAFETY: `process` is a live handle that this Watch owns. There is no further fallback if this fails too. + let _ = unsafe { windows::TerminateProcess(w.process, 1) }; + } + } +} + +#[cfg(not(any( + target_os = "macos", + target_os = "linux", + target_os = "android", + windows +)))] +mod os { + use super::Watch; + pub(super) fn tree_usage(_: &Watch) -> u64 { + 0 + } + pub(super) fn kill_tree(w: &Watch) { + // SAFETY: kill(2) has no memory-safety preconditions. + unsafe { libc::kill(w.pid, w.signal as core::ffi::c_int) }; + } +} diff --git a/src/spawn/process.rs b/src/spawn/process.rs index 3a77a8a30883..9dc8b5435587 100644 --- a/src/spawn/process.rs +++ b/src/spawn/process.rs @@ -680,6 +680,14 @@ impl Process { self.exit_handler = ProcessExitHandler::default(); } + #[cfg(windows)] + pub fn os_handle(&self) -> Option { + match &self.poller { + Poller::Uv(handle) => Some(handle.process_handle), + _ => None, + } + } + pub fn kill(&mut self, signal: u8) -> Maybe<()> { #[cfg(unix)] { diff --git a/src/sys/windows/mod.rs b/src/sys/windows/mod.rs index 594bfda0a2cc..aa86676d13d4 100644 --- a/src/sys/windows/mod.rs +++ b/src/sys/windows/mod.rs @@ -559,6 +559,9 @@ pub use bun_windows_sys::{PEB, RTL_USER_PROCESS_PARAMETERS, TEB, teb}; pub use bun_windows_sys::externs::CreateJobObjectA; pub use bun_windows_sys::externs::AssignProcessToJobObject; +pub use bun_windows_sys::externs::QueryInformationJobObject; +pub use bun_windows_sys::externs::TerminateJobObject; +pub use bun_windows_sys::externs::TerminateProcess; pub use bun_windows_sys::externs::GetCurrentProcess; @@ -571,9 +574,11 @@ pub use bun_windows_sys::externs::{ // live in bun_windows_sys::externs; Zeroable impls for these nominal types // live in bun_core/lib.rs (orphan-rule home). Do NOT re-declare here. pub use bun_windows_sys::externs::{ - IO_COUNTERS, JOBOBJECT_ASSOCIATE_COMPLETION_PORT, JOBOBJECT_BASIC_LIMIT_INFORMATION, - JOBOBJECT_EXTENDED_LIMIT_INFORMATION, JobObjectAssociateCompletionPortInformation, - JobObjectExtendedLimitInformation, + IO_COUNTERS, JOB_OBJECT_LIMIT_JOB_MEMORY, JOB_OBJECT_MSG_ACTIVE_PROCESS_ZERO, + JOB_OBJECT_MSG_NOTIFICATION_LIMIT, JOBOBJECT_ASSOCIATE_COMPLETION_PORT, + JOBOBJECT_BASIC_LIMIT_INFORMATION, JOBOBJECT_EXTENDED_LIMIT_INFORMATION, + JOBOBJECT_NOTIFICATION_LIMIT_INFORMATION, JobObjectAssociateCompletionPortInformation, + JobObjectExtendedLimitInformation, JobObjectNotificationLimitInformation, }; pub use bun_windows_sys::externs::SetInformationJobObject; @@ -1428,7 +1433,7 @@ pub struct PROCESS_MEMORY_COUNTERS { pub(crate) QuotaPagedPoolUsage: usize, pub(crate) QuotaPeakNonPagedPoolUsage: usize, pub(crate) QuotaNonPagedPoolUsage: usize, - pub(crate) PagefileUsage: usize, + pub PagefileUsage: usize, pub(crate) PeakPagefileUsage: usize, } @@ -1454,6 +1459,69 @@ pub fn GetProcessMemoryInfo(process: HANDLE) -> Result (Option, u64) { + use bun_windows_sys::externs::{ + JOBOBJECT_MEMORY_USAGE_INFORMATION, JobObjectMemoryUsageInformation, + }; + let mut usage = JOBOBJECT_MEMORY_USAGE_INFORMATION::default(); + // SAFETY: out-buffer sized for the info class. + if unsafe { + QueryInformationJobObject( + job, + JobObjectMemoryUsageInformation, + (&raw mut usage).cast(), + size_of::() as DWORD, + core::ptr::null_mut(), + ) + } != 0 + { + return (Some(usage.JobMemory), usage.PeakJobMemoryUsed); + } + let mut ext: JOBOBJECT_EXTENDED_LIMIT_INFORMATION = bun_core::ffi::zeroed(); + // SAFETY: out-buffer sized for the info class. + if unsafe { + QueryInformationJobObject( + job, + JobObjectExtendedLimitInformation, + (&raw mut ext).cast(), + size_of::() as DWORD, + core::ptr::null_mut(), + ) + } != 0 + { + return (None, ext.PeakJobMemoryUsed as u64); + } + (None, 0) +} + +/// Ask the kernel to post `JOB_OBJECT_MSG_NOTIFICATION_LIMIT` with `key` to `port` when `job` commits more than `limit` bytes. +pub fn job_notify_memory_limit(job: HANDLE, port: HANDLE, key: usize, limit: u64) -> bool { + let mut assoc = JOBOBJECT_ASSOCIATE_COMPLETION_PORT { + CompletionKey: key as LPVOID, + CompletionPort: port, + }; + let mut notify = JOBOBJECT_NOTIFICATION_LIMIT_INFORMATION { + JobMemoryLimit: limit, + LimitFlags: JOB_OBJECT_LIMIT_JOB_MEMORY, + ..Default::default() + }; + // SAFETY: both in-buffers are sized for their info class; bad handles fail with 0. + unsafe { + SetInformationJobObject( + job, + JobObjectAssociateCompletionPortInformation, + (&raw mut assoc).cast(), + size_of::() as DWORD, + ) != 0 + && SetInformationJobObject( + job, + JobObjectNotificationLimitInformation, + (&raw mut notify).cast(), + size_of::() as DWORD, + ) != 0 + } +} pub use bun_windows_sys::externs::GetConsoleMode; pub use bun_windows_sys::externs::SetConsoleMode; diff --git a/src/windows_sys/externs.rs b/src/windows_sys/externs.rs index b84cbfd3a9bf..382fa23aa7e1 100644 --- a/src/windows_sys/externs.rs +++ b/src/windows_sys/externs.rs @@ -1444,6 +1444,18 @@ unsafe extern "system" { ) -> BOOL; pub fn OpenProcess(dwDesiredAccess: DWORD, bInheritHandle: BOOL, dwProcessId: DWORD) -> HANDLE; + + pub fn QueryInformationJobObject( + hJob: HANDLE, + JobObjectInformationClass: DWORD, + lpJobObjectInformation: LPVOID, + cbJobObjectInformationLength: DWORD, + lpReturnLength: LPDWORD, + ) -> BOOL; + + pub fn TerminateJobObject(hJob: HANDLE, uExitCode: UINT) -> BOOL; + + pub fn TerminateProcess(hProcess: HANDLE, uExitCode: UINT) -> BOOL; } unsafe extern "C" { @@ -1459,6 +1471,38 @@ unsafe extern "C" { pub const JobObjectAssociateCompletionPortInformation: DWORD = 7; /// `JOBOBJECTINFOCLASS::JobObjectExtendedLimitInformation` (`winnt.h`). pub const JobObjectExtendedLimitInformation: DWORD = 9; +/// `JOBOBJECTINFOCLASS::JobObjectNotificationLimitInformation` (`winnt.h`, Windows 8+). +pub const JobObjectNotificationLimitInformation: DWORD = 12; +/// `JOBOBJECT_NOTIFICATION_LIMIT_INFORMATION.LimitFlags` bit that arms `JobMemoryLimit`. +pub const JOB_OBJECT_LIMIT_JOB_MEMORY: DWORD = 0x0000_0200; +/// Completion port message: the last process in the job exited. +pub const JOB_OBJECT_MSG_ACTIVE_PROCESS_ZERO: DWORD = 4; +/// Completion port message: the job crossed a notification limit. Nothing fails inside the job. +pub const JOB_OBJECT_MSG_NOTIFICATION_LIMIT: DWORD = 11; + +/// `JOBOBJECT_NOTIFICATION_LIMIT_INFORMATION` (`winnt.h`). +#[repr(C)] +#[derive(Clone, Copy, Default)] +pub struct JOBOBJECT_NOTIFICATION_LIMIT_INFORMATION { + pub IoReadBytesLimit: u64, + pub IoWriteBytesLimit: u64, + pub PerJobUserTimeLimit: i64, + pub JobMemoryLimit: u64, + pub RateControlTolerance: DWORD, + pub RateControlToleranceInterval: DWORD, + pub LimitFlags: DWORD, +} + +/// `JOBOBJECTINFOCLASS::JobObjectMemoryUsageInformation` (`winnt.h`, Windows 10 1607+). +pub const JobObjectMemoryUsageInformation: DWORD = 28; + +/// `JOBOBJECT_MEMORY_USAGE_INFORMATION` (`winnt.h`). +#[repr(C)] +#[derive(Clone, Copy, Default)] +pub struct JOBOBJECT_MEMORY_USAGE_INFORMATION { + pub JobMemory: u64, + pub PeakJobMemoryUsed: u64, +} /// `WAITORTIMERCALLBACK` (`winnt.h`) — thread-pool callback for /// `RegisterWaitForSingleObject`. `TimerOrWaitFired` is `TRUE` on timeout. diff --git a/test/integration/bun-types/fixture/spawn.ts b/test/integration/bun-types/fixture/spawn.ts index 2036413158a0..6d4b0408f72c 100644 --- a/test/integration/bun-types/fixture/spawn.ts +++ b/test/integration/bun-types/fixture/spawn.ts @@ -226,3 +226,14 @@ tsd.expectAssignable(); + tsd.expectType(limited.exitedDueToMaxMemory).is(); + tsd.expectType(Bun.spawnSync(["echo", "hi"], { maxMemory: 1024 }).exitedDueToMaxMemory).is(); + nodeSpawn("echo", ["hi"], { maxMemory: 1024 }); + nodeSpawnSync("echo", ["hi"], { maxMemory: 1024 }); +} diff --git a/test/js/bun/spawn/spawn-max-memory.test.ts b/test/js/bun/spawn/spawn-max-memory.test.ts new file mode 100644 index 000000000000..6b15aaad93d8 --- /dev/null +++ b/test/js/bun/spawn/spawn-max-memory.test.ts @@ -0,0 +1,396 @@ +import { subprocessInternals } from "bun:internal-for-testing"; +import { beforeAll, describe, expect, test } from "bun:test"; +import { bunEnv, bunExe, isLinux, isMacOS, isWindows } from "harness"; +import { spawn as cpSpawn, spawnSync as cpSpawnSync } from "node:child_process"; +import { mkdirSync, readFileSync, rmdirSync, writeFileSync } from "node:fs"; +import { join } from "node:path"; + +const MB = 1024 * 1024; + +// Touch every page so the memory is resident, then park. +const hog = (mb: number) => + `const b = Buffer.allocUnsafe(${mb} * 1024 * 1024); for (let i = 0; i < b.length; i += 4096) b[i] = 1; globalThis.keep = b; setInterval(() => {}, 1000);`; + +// An idle debug build uses far more memory than a release build, so every limit comes from a measured idle size. +let idleBytes = 0; +beforeAll(async () => { + await using idle = Bun.spawn({ + cmd: [bunExe(), "-e", `console.log("ready"); setInterval(() => {}, 1000);`], + env: bunEnv, + stdio: ["ignore", "pipe", "ignore"], + }); + for await (const chunk of idle.stdout) { + if (Buffer.from(chunk).includes("ready")) break; + } + idleBytes = idle.memoryUsage().current; + expect(idleBytes).toBeGreaterThan(0); +}); + +// A limit that a tree of `processes` idle Bun processes stays well below. +const limitFor = (processes: number) => Math.ceil((processes * idleBytes * 1.5) / MB) * MB + 64 * MB; +// A program that goes well over that limit. +const hogFor = (processes: number) => hog(limitFor(processes) / MB + 256); + +// A killed process stays visible as a zombie until its new parent reaps it, so wait for the pid to go away. +async function gone(pid: number) { + for (;;) { + try { + process.kill(pid, 0); + } catch { + return; + } + await Bun.sleep(5); + } +} + +// True when this host lets us create a memory cgroup inside our own, which is when Bun must pick the cgroup route. +function canCreateMemoryCgroup(): boolean { + if (!isLinux) return false; + const name = `bun-max-memory-probe-${process.pid}`; + const candidates: { dir: string; v2: boolean }[] = []; + for (const line of readFileSync("/proc/self/cgroup", "utf8").split("\n")) { + const [, controllers, ...rest] = line.split(":"); + if (controllers === undefined) continue; + const own = rest.join(":").replace(/\/$/, ""); + // v1 lets a populated cgroup have limited children. v2 allows that only in the real root cgroup. + if (controllers.split(",").includes("memory")) + candidates.push({ dir: `/sys/fs/cgroup/memory${own}/${name}`, v2: false }); + else if (controllers === "" && own === "") candidates.push({ dir: `/sys/fs/cgroup/${name}`, v2: true }); + } + let hasSwap = false; + try { + hasSwap = readFileSync("/proc/swaps", "utf8").trim().split("\n").length > 1; + } catch {} + for (const { dir, v2 } of candidates) { + try { + mkdirSync(dir); + } catch { + continue; + } + try { + // "r+" never creates a file. On a tmpfs that is not a cgroup mount, the limit files do not exist, as for Bun. + writeFileSync(join(dir, v2 ? "memory.max" : "memory.limit_in_bytes"), String(1024 * MB), { flag: "r+" }); + // Bun does not use a v1 cgroup that cannot report its OOM kills (before Linux 4.13) or that has the OOM killer off. + if (!v2) { + const control = readFileSync(join(dir, "memory.oom_control"), "utf8"); + if (!/^oom_kill /m.test(control) || /^oom_kill_disable 1/m.test(control)) continue; + } + try { + writeFileSync(join(dir, v2 ? "memory.swap.max" : "memory.memsw.limit_in_bytes"), v2 ? "0" : String(1024 * MB), { + flag: "r+", + }); + } catch { + if (hasSwap) continue; + } + return true; + } catch { + } finally { + try { + rmdirSync(dir); + } catch {} + } + } + return false; +} + +// On such a host Bun must pick the cgroup route, and the tests that need the sampler must skip. +const hasMemoryCgroup = canCreateMemoryCgroup(); + +describe("Bun.spawn maxMemory", () => { + test("the kernel enforces the limit where it can", async () => { + const idle = [bunExe(), "-e", "setInterval(() => {}, 1000)"]; + await using limited = Bun.spawn({ + cmd: idle, + env: bunEnv, + stdio: ["ignore", "ignore", "ignore"], + maxMemory: 1024 * MB, + }); + const expected = isWindows ? "job" : isMacOS ? "sampler" : hasMemoryCgroup ? "cgroup" : "sampler"; + expect(subprocessInternals.memoryLimitRoute(limited)).toBe(expected); + + await using unlimited = Bun.spawn({ cmd: idle, env: bunEnv, stdio: ["ignore", "ignore", "ignore"] }); + expect(subprocessInternals.memoryLimitRoute(unlimited)).toBeUndefined(); + }); + + test.concurrent("kills a child that exceeds the limit", async () => { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", hogFor(1)], + env: bunEnv, + stdio: ["ignore", "ignore", "ignore"], + maxMemory: limitFor(1), + killSignal: "SIGKILL", + }); + await proc.exited; + if (!isWindows) expect(proc.signalCode).toBe("SIGKILL"); + expect(proc.exitCode === 0).toBe(false); + }); + + test.concurrent("leaves a child under the limit alone", async () => { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", `Buffer.alloc(8 * 1024 * 1024, 1); console.log("ok")`], + env: bunEnv, + stdio: ["ignore", "pipe", "ignore"], + maxMemory: 1024 * MB, + }); + const [stdout, exitCode] = await Promise.all([proc.stdout.text(), proc.exited]); + expect(stdout).toBe("ok\n"); + expect(proc.signalCode).toBeNull(); + expect(exitCode).toBe(0); + }); + + test.concurrent("counts and kills grandchildren", async () => { + // Parent is small; the grandchild is the hog. The limit applies to the tree. + const parent = ` + const child = Bun.spawn({ cmd: [process.execPath, "-e", ${JSON.stringify(hogFor(2))}], stdio: ["ignore", "ignore", "ignore"] }); + console.log(child.pid); + await child.exited; + console.log("grandchild exited", child.signalCode ?? child.exitCode); + setInterval(() => {}, 1000); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", parent], + env: bunEnv, + stdio: ["ignore", "pipe", "inherit"], + maxMemory: limitFor(2), + killSignal: "SIGKILL", + }); + const [stdout] = await Promise.all([proc.stdout.text(), proc.exited]); + const grandchildPid = parseInt(stdout.split("\n")[0], 10); + expect(grandchildPid).toBeGreaterThan(0); + if (!isWindows) expect(proc.signalCode).toBe("SIGKILL"); + // The grandchild must be gone too, not just the direct child. The test times out if it survives. + await gone(grandchildPid); + }); + + test.concurrent("honors killSignal", async () => { + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", hogFor(1)], + env: bunEnv, + stdio: ["ignore", "ignore", "ignore"], + maxMemory: limitFor(1), + killSignal: "SIGTERM", + }); + await proc.exited; + // In a Linux cgroup the kernel does the kill, and it always uses SIGKILL. + if (isLinux) expect(["SIGTERM", "SIGKILL"]).toContain(proc.signalCode); + else if (!isWindows) expect(proc.signalCode).toBe("SIGTERM"); + }); + + test.concurrent("memoryUsage() reports the tree, with and without maxMemory", async () => { + for (const maxMemory of [undefined, 2048 * MB]) { + // The child is small. The grandchild holds 128 MB, so a tree total proves descendants count. + // The grandchild exits when its stdin closes, which happens when the child dies. + const grandchild = hog(128) + ` process.stdin.on("close", () => process.exit(0)).resume(); console.log("ready");`; + const parent = ` + const child = Bun.spawn({ cmd: [process.execPath, "-e", ${JSON.stringify(grandchild)}], stdio: ["pipe", "pipe", "ignore"] }); + for await (const chunk of child.stdout) { if (Buffer.from(chunk).includes("ready")) break; } + console.log("ready " + child.pid); + setInterval(() => {}, 1000); + `; + await using proc = Bun.spawn({ + cmd: [bunExe(), "-e", parent], + env: bunEnv, + stdio: ["ignore", "pipe", "inherit"], + maxMemory, + }); + let output = ""; + for await (const chunk of proc.stdout) { + output += Buffer.from(chunk).toString(); + if (output.includes("\n")) break; + } + const grandchildPid = parseInt(output.split(" ")[1], 10); + expect(grandchildPid).toBeGreaterThan(0); + const usage = proc.memoryUsage(); + // Windows counts only the root process when there is no Job Object. + if (!isWindows || maxMemory) expect(usage.current).toBeGreaterThan(128 * MB); + expect(usage.peak).toBeGreaterThanOrEqual(usage.current); + + proc.kill("SIGKILL"); + await proc.exited; + const after = proc.memoryUsage(); + expect(after.current).toBe(0); + expect(after.peak).toBeGreaterThanOrEqual(usage.peak); + + await gone(grandchildPid); + } + }); + + test("spawnSync reports exitedDueToMaxMemory", () => { + const over = Bun.spawnSync({ + cmd: [bunExe(), "-e", hogFor(1)], + env: bunEnv, + stdio: ["ignore", "ignore", "ignore"], + maxMemory: limitFor(1), + killSignal: "SIGKILL", + }); + expect(over.exitedDueToMaxMemory).toBe(true); + expect(over.success).toBe(false); + + const under = Bun.spawnSync({ + cmd: [bunExe(), "-e", "1"], + env: bunEnv, + stdio: ["ignore", "ignore", "ignore"], + maxMemory: 1024 * MB, + }); + expect(under.exitedDueToMaxMemory).toBe(false); + expect(under.exitCode).toBe(0); + }); + + test("validates the option", () => { + expect(() => Bun.spawn({ cmd: [bunExe(), "-e", "1"], maxMemory: -1 })).toThrow(RangeError); + expect(() => Bun.spawn({ cmd: [bunExe(), "-e", "1"], maxMemory: 1.5 })).toThrow(TypeError); + // NaN must not turn the limit off without an error. + expect(() => Bun.spawn({ cmd: [bunExe(), "-e", "1"], maxMemory: NaN })).toThrow(RangeError); + // 0, null, undefined and Infinity all mean "no limit". + for (const maxMemory of [0, null, undefined, Infinity]) { + const r = Bun.spawnSync({ cmd: [bunExe(), "-e", "1"], env: bunEnv, maxMemory: maxMemory as any }); + expect(r.exitCode).toBe(0); + } + }); + + test.concurrent("exitedDueToMaxMemory tells a memory kill from a normal exit", async () => { + await using over = Bun.spawn({ + cmd: [bunExe(), "-e", hogFor(1)], + env: bunEnv, + stdio: ["ignore", "ignore", "ignore"], + maxMemory: limitFor(1), + killSignal: "SIGKILL", + }); + await over.exited; + expect(over.exitedDueToMaxMemory).toBe(true); + + await using under = Bun.spawn({ + cmd: [bunExe(), "-e", "1"], + env: bunEnv, + stdio: ["ignore", "ignore", "ignore"], + maxMemory: 1024 * MB, + }); + expect(await under.exited).toBe(0); + expect(under.exitedDueToMaxMemory).toBe(false); + }); + + // On Windows a child dies with its parent through libuv's kill-on-close job, so a descendant is never reparented there. + (isWindows ? test.skip : test.concurrent)( + "a descendant whose parent exited is still counted and killed", + async () => { + // root sh -> middle sh -> hog. The hog gets the middle pid and allocates only when that sh is no longer its parent. + const hogSrc = `const middle = Number(process.argv.at(-1)); const timer = setInterval(() => { if (process.ppid !== middle) { clearInterval(timer); ${hogFor(1)} } }, 5);`; + const middle = `"$0" -e "$1" $$ >/dev/null 2>&1 & echo $!; read _`; + await using proc = Bun.spawn({ + cmd: ["sh", "-c", `sh -c '${middle}' "$0" "$1"; exec sleep 100000`, bunExe(), hogSrc], + // The ASAN CI lanes set BUN_FEATURE_FLAG_NO_ORPHANS, which kills the hog the moment its parent sh exits. + env: { ...bunEnv, BUN_FEATURE_FLAG_NO_ORPHANS: undefined }, + stdio: ["pipe", "pipe", "inherit"], + maxMemory: limitFor(1), + killSignal: "SIGKILL", + }); + let output = ""; + for await (const chunk of proc.stdout) { + output += Buffer.from(chunk).toString(); + if (output.includes("\n")) break; + } + const hogPid = parseInt(output, 10); + expect(hogPid).toBeGreaterThan(0); + // This walk records the hog as a member while its parent is still alive. + expect(proc.memoryUsage().current).toBeGreaterThan(0); + + proc.stdin.write("\n"); + await proc.stdin.flush(); + await proc.exited; + expect(proc.exitedDueToMaxMemory).toBe(true); + await gone(hogPid); + }, + ); + // A Windows job and a Linux cgroup stop every process at once, so only the sampler can miss a late process. + (isWindows || hasMemoryCgroup ? test.skip : test.concurrent)( + "a process that starts after the kill is signalled too", + async () => { + // Both processes exit when their stdin closes, so nothing outlives a failed run. + const lateChild = `process.stdin.on("close", () => process.exit(0)).resume();`; + const rootSrc = ` + process.stdin.on("close", () => process.exit(0)).resume(); + process.on("SIGTERM", () => { + const late = Bun.spawn({ cmd: [process.execPath, "-e", ${JSON.stringify(lateChild)}], stdio: ["pipe", "ignore", "ignore"] }); + console.log(late.pid); + }); + ${hogFor(2)} + `; + const proc = Bun.spawn({ + cmd: [bunExe(), "-e", rootSrc], + env: bunEnv, + stdio: ["pipe", "pipe", "inherit"], + maxMemory: limitFor(2), + killSignal: "SIGTERM", + }); + try { + expect(subprocessInternals.memoryLimitRoute(proc)).toBe("sampler"); + let output = ""; + for await (const chunk of proc.stdout) { + output += Buffer.from(chunk).toString(); + if (output.includes("\n")) break; + } + const latePid = parseInt(output, 10); + expect(latePid).toBeGreaterThan(0); + // The root ignores SIGTERM and stays alive. The late child does not, so it dies only if a later sample signals it. + await gone(latePid); + } finally { + proc.kill("SIGKILL"); + await proc.exited; + } + }, + ); +}); + +describe("node:child_process maxMemory", () => { + test.concurrent("a lower-case killSignal still works, with and without maxMemory", async () => { + for (const maxMemory of [undefined, 1024 * MB]) { + const child = cpSpawn(bunExe(), ["-e", "setInterval(() => {}, 1000)"], { + env: bunEnv, + killSignal: "sigterm", + maxMemory, + }); + const { promise, resolve, reject } = Promise.withResolvers(); + child.on("error", reject); + child.on("exit", (_, signal) => resolve(signal)); + child.on("spawn", () => child.kill()); + const signal = await promise; + if (!isWindows) expect(signal).toBe("SIGTERM"); + } + }); + + test.concurrent("spawn kills a child that exceeds the limit", async () => { + const child = cpSpawn(bunExe(), ["-e", hogFor(1)], { + env: bunEnv, + stdio: "ignore", + maxMemory: limitFor(1), + killSignal: "SIGKILL", + }); + const { promise, resolve, reject } = Promise.withResolvers<[number | null, NodeJS.Signals | null]>(); + child.on("error", reject); + child.on("exit", (code, signal) => resolve([code, signal])); + const [code, signal] = await promise; + if (isWindows) expect(code).not.toBe(0); + else expect(signal).toBe("SIGKILL"); + }); + + test("spawnSync reports ENOMEM", () => { + const result = cpSpawnSync(bunExe(), ["-e", hogFor(1)], { + env: bunEnv, + stdio: "ignore", + maxMemory: limitFor(1), + killSignal: "SIGKILL", + }); + expect((result.error as NodeJS.ErrnoException)?.code).toBe("ENOMEM"); + }); + + test("an invalid maxMemory throws before anything is spawned", () => { + for (const maxMemory of [-1, 1.5, NaN, "64"]) { + expect(() => cpSpawn(bunExe(), ["-e", "1"], { maxMemory: maxMemory as any })).toThrow( + expect.objectContaining({ code: "ERR_OUT_OF_RANGE" }), + ); + expect(() => cpSpawnSync(bunExe(), ["-e", "1"], { maxMemory: maxMemory as any })).toThrow( + expect.objectContaining({ code: "ERR_OUT_OF_RANGE" }), + ); + } + }); +});