diff --git a/src/runtime/shell/Builtin.rs b/src/runtime/shell/Builtin.rs index 6ce494deb4ed..a1ca5b9817df 100644 --- a/src/runtime/shell/Builtin.rs +++ b/src/runtime/shell/Builtin.rs @@ -63,6 +63,21 @@ pub(crate) trait BuiltinState: Sized { fn state_mut(interp: &Interpreter, cmd: NodeId) -> &mut Self { Self::extract(&mut Builtin::of_mut(interp, cmd).impl_) } + + /// stdout and the state borrowed together: bytes kept on the state are enqueued uncopied. + #[inline] + #[track_caller] + fn split_stdout(bltn: &mut Builtin) -> (&mut BuiltinIO, &mut Self) { + (&mut bltn.stdout, Self::extract(&mut bltn.impl_)) + } + + /// `split_stdout` under `Builtin::write_no_io`'s contract: stdout must not `needs_io()`. + #[track_caller] + fn split_stdout_no_io(interp: &Interpreter, cmd: NodeId) -> (NoIoOutput<'_>, &mut Self) { + let (shell, bltn) = Builtin::of_mut_with_shell(interp, cmd); + let (io, state) = Self::split_stdout(bltn); + (NoIoOutput { io, shell }, state) + } } macro_rules! shell_builtins { @@ -338,16 +353,13 @@ impl BuiltinIO { } } - /// Body of [`Builtin::write_no_io`] with the Cmd split-borrow already - /// performed by the caller. Exists so builtins whose payload lives in - /// `Builtin.impl_` (disjoint from `stdout`/`stderr`) can write a borrowed - /// slice without an intermediate heap clone. + /// Body of [`NoIoOutput::write`], which pairs the stream with its `shell`. /// /// # Safety /// `shell` must point to the live `ShellExecEnv` owning this builtin /// (i.e. `cmd.base.shell`); only dereferenced for the [`BuiltinIO::Buf`] /// arm. - pub(crate) unsafe fn write_no_io_to( + unsafe fn write_no_io_to( &mut self, shell: *mut crate::shell::interpreter::ShellExecEnv, buf: &[u8], @@ -431,6 +443,21 @@ impl BuiltinIO { } } +/// A non-fd stream and the env of its own Cmd; built only in this module, so `write` can be safe. +pub(crate) struct NoIoOutput<'a> { + io: &'a mut BuiltinIO, + shell: *mut crate::shell::interpreter::ShellExecEnv, +} + +impl NoIoOutput<'_> { + /// Returns `Err(ENOSPC)` when an ArrayBuffer target is already full. + pub(crate) fn write(&mut self, buf: &[u8]) -> bun_sys::Result { + // SAFETY: `shell` is the env of the Cmd `io` is borrowed from; the env + // outlives the Cmd, which stays borrowed through `io`. + unsafe { self.io.write_no_io_to(self.shell, buf) } + } +} + impl BuiltinInput { fn from_in_kind(ik: &InKind) -> BuiltinInput { match ik { @@ -880,8 +907,22 @@ impl Builtin { #[inline] #[track_caller] pub(crate) fn of_mut<'a>(interp: &'a Interpreter, cmd: NodeId) -> &'a mut Builtin { - match &mut interp.as_cmd_mut(cmd).exec { - crate::shell::states::cmd::Exec::Builtin(b) => b, + Self::of_mut_with_shell(interp, cmd).1 + } + + /// [`of_mut`](Self::of_mut) plus the Cmd's shell env, the pair a [`NoIoOutput`] needs. + #[inline] + #[track_caller] + fn of_mut_with_shell<'a>( + interp: &'a Interpreter, + cmd: NodeId, + ) -> ( + *mut crate::shell::interpreter::ShellExecEnv, + &'a mut Builtin, + ) { + let cmd_node = interp.as_cmd_mut(cmd); + match &mut cmd_node.exec { + crate::shell::states::cmd::Exec::Builtin(b) => (cmd_node.base.shell, &mut **b), _ => panic!("Cmd {} is not running a builtin", cmd), } } @@ -915,20 +956,13 @@ impl Builtin { if buf.is_empty() { return Ok(0); } - // Split-borrow the Cmd so `shell` - // and the builtin's stdout/stderr are accessible simultaneously. - let cmd_node = interp.as_cmd_mut(cmd); - let shell = cmd_node.base.shell; - let crate::shell::states::cmd::Exec::Builtin(me) = &mut cmd_node.exec else { - panic!("Cmd {} is not running a builtin", cmd); - }; - let out: &mut BuiltinIO = match io_kind { + let (shell, me) = Self::of_mut_with_shell(interp, cmd); + let io = match io_kind { IoKind::Stdout => &mut me.stdout, IoKind::Stderr => &mut me.stderr, IoKind::Stdin => return Ok(0), }; - // SAFETY: `shell` is `cmd_node.base.shell`, live for the Cmd's lifetime. - unsafe { out.write_no_io_to(shell, buf) } + NoIoOutput { io, shell }.write(buf) } /// Shell exec env of the owning Cmd. diff --git a/src/runtime/shell/builtin/seq.rs b/src/runtime/shell/builtin/seq.rs index fbad2252f12f..41b09a4b528e 100644 --- a/src/runtime/shell/builtin/seq.rs +++ b/src/runtime/shell/builtin/seq.rs @@ -1,15 +1,21 @@ use std::io::Write as _; -use crate::shell::builtin::{Builtin, BuiltinState, IoKind, Kind}; -use crate::shell::interpreter::{Interpreter, NodeId}; +use crate::shell::builtin::{Builtin, BuiltinState, Kind}; +use crate::shell::interpreter::{Interpreter, NodeId, OutputNeedsIOSafeGuard}; use crate::shell::io_writer::{ChildPtr, WriterTag}; use crate::shell::yield_::Yield; +/// Chunks are cut at the first value boundary at or past this size; about one is held at a time. +const CHUNK_SIZE: usize = 64 * 1024; + #[derive(Clone, Copy, PartialEq, Eq, Default)] enum State { #[default] Idle, + /// A chunk is being written to stdout and more values follow it. + Writing, Err, + /// The chunk being written (if any) is the last one. Done, } @@ -18,6 +24,10 @@ pub struct Seq { start: f32, end: f32, increment: f32, + /// Next value to render. + current: f32, + /// The chunk currently being written; reused for every chunk. + buf: Vec, /// Borrowed from argv (NUL-terminated arena strings) or `'static` literals; /// argv outlives the builtin — `RawSlice` invariant. separator: bun_ptr::RawSlice, @@ -31,6 +41,8 @@ impl Default for Seq { start: 1.0, end: 1.0, increment: 1.0, + current: 1.0, + buf: Vec::new(), separator: bun_ptr::RawSlice::new(b"\n"), terminator: bun_ptr::RawSlice::EMPTY, } @@ -156,46 +168,73 @@ impl Seq { } fn do_(interp: &Interpreter, cmd: NodeId) -> Yield { - let needs_io = Builtin::of(interp, cmd).stdout.needs_io().is_some(); - // Render entirely into a local Vec, then either enqueue it or - // write_no_io it; we buffer once for simplicity. - let (start, end, incr, sep, term) = { + { let me = Self::state_mut(interp, cmd); - (me.start, me.end, me.increment, me.separator, me.terminator) + me.current = me.start; + } + if let Some(safeguard) = Builtin::of(interp, cmd).stdout.needs_io() { + return Self::enqueue_chunk(interp, cmd, safeguard); + } + loop { + let (mut stdout, me) = Self::split_stdout_no_io(interp, cmd); + let last = me.render_chunk(); + // Err: the `> ${buffer}` is full, so no later chunk would fit either. + let written = stdout.write(&me.buf); + if last || written.is_err() { + break; + } + } + Self::state_mut(interp, cmd).state = State::Done; + Builtin::done(interp, cmd, 0) + } + + /// Queues the next chunk; `on_io_writer_chunk` queues the one after it. + fn enqueue_chunk( + interp: &Interpreter, + cmd: NodeId, + safeguard: OutputNeedsIOSafeGuard, + ) -> Yield { + let child = ChildPtr::new(cmd, WriterTag::Builtin); + let (stdout, me) = Self::split_stdout(Builtin::of_mut(interp, cmd)); + me.state = if me.render_chunk() { + State::Done + } else { + State::Writing }; - let mut out = Vec::new(); - let mut current = start; - while if incr > 0.0 { - current <= end + stdout.enqueue(child, &me.buf, safeguard) + } + + fn has_next(&self) -> bool { + if self.increment > 0.0 { + self.current <= self.end } else { - current >= end - } { + self.current >= self.end + } + } + + /// Refills `buf`; true once the sequence (and terminator) has been rendered into it. + fn render_chunk(&mut self) -> bool { + self.buf.clear(); + while self.has_next() { + if self.buf.len() >= CHUNK_SIZE { + return false; + } // Rust `{}` for f32 prints the shortest decimal that round-trips // (no exponent, no trailing ".0"). - let _ = write!(&mut out, "{}", current); - out.extend_from_slice(sep.slice()); - let next = current + incr; - if next == current { + let _ = write!(&mut self.buf, "{}", self.current); + self.buf.extend_from_slice(self.separator.slice()); + let next = self.current + self.increment; + if next == self.current { // f32 rounding can make `current + incr` equal `current` // (e.g. `seq 1 99999999` saturates at 2^24, or a tiny // increment relative to `current`). Without this check the - // loop never terminates and `out` grows without bound. + // sequence would never end. break; } - current = next; - } - out.extend_from_slice(term.slice()); - - Self::state_mut(interp, cmd).state = State::Done; - if needs_io { - let safeguard = Builtin::of(interp, cmd).stdout.needs_io().unwrap(); - let child = ChildPtr::new(cmd, WriterTag::Builtin); - return Builtin::of_mut(interp, cmd) - .stdout - .enqueue(child, &out, safeguard); + self.current = next; } - let _ = Builtin::write_no_io(interp, cmd, IoKind::Stdout, &out); - Builtin::done(interp, cmd, 0) + self.buf.extend_from_slice(self.terminator.slice()); + true } pub(crate) fn on_io_writer_chunk( @@ -209,6 +248,10 @@ impl Seq { return Builtin::done(interp, cmd, 1); } match Self::state_mut(interp, cmd).state { + State::Writing => { + debug_assert!(Builtin::of(interp, cmd).stdout.needs_io().is_some()); + Self::enqueue_chunk(interp, cmd, OutputNeedsIOSafeGuard::OutputNeedsIo) + } State::Done => Builtin::done(interp, cmd, 0), State::Err => Builtin::done(interp, cmd, 1), State::Idle => { diff --git a/src/runtime/shell/builtin/yes.rs b/src/runtime/shell/builtin/yes.rs index 5b97cb46616b..416be25bfdfd 100644 --- a/src/runtime/shell/builtin/yes.rs +++ b/src/runtime/shell/builtin/yes.rs @@ -1,8 +1,7 @@ use crate::shell::ExitCode; -use crate::shell::builtin::{Builtin, BuiltinIO, BuiltinState, Impl, Kind}; +use crate::shell::builtin::{Builtin, BuiltinState, Kind}; use crate::shell::interpreter::{EventLoopHandle, Interpreter, NodeId, OutputNeedsIOSafeGuard}; use crate::shell::io_writer::{ChildPtr, WriterTag}; -use crate::shell::states::cmd::Exec; use crate::shell::yield_::Yield; use bun_event_loop::ConcurrentTask::AutoDeinit; @@ -89,21 +88,12 @@ impl Yes { /// Write 4 chunks then bounce to the event loop so we don't hog the main /// thread. fn write_no_io_loop(interp: &Interpreter, cmd: NodeId) -> Yield { - // Split-borrow the Cmd so the tiled buffer (in `impl_`) and `stdout` - // are accessible simultaneously — the buffer is written zero-copy, - // which matters for `yes` throughput. let err = { - let cmd_node = interp.as_cmd_mut(cmd); - let shell = cmd_node.base.shell; - let Exec::Builtin(me) = &mut cmd_node.exec else { - unreachable!() - }; - let (stdout, yes) = Self::split_stdout_state(me); + let (mut stdout, yes) = Self::split_stdout_no_io(interp, cmd); let chunk = &yes.buffer[..yes.buffer_used]; let mut err = None; for _ in 0..4 { - // SAFETY: `shell` is `cmd_node.base.shell`, live for the Cmd. - if let Err(e) = unsafe { stdout.write_no_io_to(shell, chunk) } { + if let Err(e) = stdout.write(chunk) { err = Some(e); break; } @@ -140,9 +130,7 @@ impl Yes { safeguard: OutputNeedsIOSafeGuard, ) -> Yield { let child = ChildPtr::new(cmd, WriterTag::Builtin); - // `stdout` and `impl_` are disjoint fields of `Builtin` — split-borrow - // so the tiled buffer is enqueued zero-copy. - let (stdout, yes) = Self::split_stdout_state(Builtin::of_mut(interp, cmd)); + let (stdout, yes) = Self::split_stdout(Builtin::of_mut(interp, cmd)); stdout.enqueue(child, &yes.buffer[..yes.buffer_used], safeguard) } @@ -172,16 +160,6 @@ impl Yes { debug_assert!(Builtin::of(interp, cmd).stdout.needs_io().is_some()); Self::enqueue_chunk(interp, cmd, OutputNeedsIOSafeGuard::OutputNeedsIo) } - - /// Split-borrow `&mut Builtin` into `(&mut stdout, &mut Yes)`; the fields - /// are disjoint so this is a sound reborrow without `unsafe`. - #[inline] - fn split_stdout_state(me: &mut Builtin) -> (&mut BuiltinIO, &mut Yes) { - let Impl::Yes(yes) = &mut me.impl_ else { - unreachable!() - }; - (&mut me.stdout, &mut **yes) - } } // `buffer: Vec` drops with the owning `Box`; no explicit `Drop` impl diff --git a/test/js/bun/shell/commands/seq.test.ts b/test/js/bun/shell/commands/seq.test.ts index 5ff6b24aefbc..743ca256bec4 100644 --- a/test/js/bun/shell/commands/seq.test.ts +++ b/test/js/bun/shell/commands/seq.test.ts @@ -1,5 +1,7 @@ +import { $ } from "bun"; import { describe, expect, test } from "bun:test"; -import { bunEnv, bunExe, isASAN } from "harness"; +import { bunEnv, bunExe, tempDir } from "harness"; +import { join } from "node:path"; import { createTestBuilder } from "../test_builder"; const TestBuilder = createTestBuilder(import.meta.path); @@ -137,38 +139,143 @@ describe("seq without stdout", async () => { .runAsTest("works basic down without stdout"); }); -// Regression guard: the fd-output path used to build the full output into a -// local Vec, store it into state, then clone the stored Vec to hand to -// BuiltinIO::enqueue (which itself copies into IOWriter's buffer). That is a -// full-output-sized clone on top of the copy that must exist, so peak RSS was -// ~3x the output instead of ~2x. ASAN-gated because release mimalloc does not -// retain freed pages the way ASAN's allocator does. -test.skipIf(!isASAN)("seq piped to an fd does not clone its output buffer before enqueue", async () => { - // 100-byte separator keeps the output large (~32 MB) with only 300k - // iterations, so the child finishes in ~1s under ASAN. - await using proc = Bun.spawn({ - cmd: [ - bunExe(), - "-e", - `const rss = process.platform === "darwin" && typeof Bun.unsafe.memoryFootprint === "function" ? Bun.unsafe.memoryFootprint : process.memoryUsage.rss;` + +// The builtin renders and writes its output about 64 KiB at a time. These +// sequences span several chunks, written to each kind of stdout that takes a +// different path through the builtin: the captured buffer (written directly), +// a file (fd written synchronously) and a pipe (fd completed from the event +// loop). +describe("seq long output", () => { + const COUNT = 40_000; + // BSD seq: the separator follows every value, the terminator comes last. + const expected = Array.from({ length: COUNT }, (_, i) => `${i + 1},`).join("") + "END"; + const copyStdin = "await Bun.write(Bun.stdout, await Bun.stdin.bytes())"; + + test.concurrent("captured stdout", async () => { + expect(await $`seq -s , -t END 1 ${COUNT}`.text()).toBe(expected); + }); + + test.concurrent("redirected to a file", async () => { + using dir = tempDir("seq-long", {}); + const out = join(String(dir), "out.txt"); + await $`seq -s , -t END 1 ${COUNT} > ${out}`.quiet(); + expect(await Bun.file(out).text()).toBe(expected); + }); + + test.concurrent("piped to a process", async () => { + const stdout = await $`seq -s , -t END 1 ${COUNT} | ${bunExe()} -e ${copyStdin}`.env(bunEnv).text(); + expect(stdout).toBe(expected); + }); + + test.concurrent("redirected to a Buffer that holds the whole sequence", async () => { + const target = Buffer.alloc(expected.length); + const { stderr, exitCode } = await $`seq -s , -t END 1 ${COUNT} > ${target}`.nothrow().quiet(); + expect({ stderr: stderr.toString(), exitCode, target: target.toString() }).toEqual({ + stderr: "", + exitCode: 0, + target: expected, + }); + }); + + // 100 KiB takes the first chunk whole and is filled up by part of the second; + // the third is refused. The Buffer must hold exactly that prefix. Like the + // other builtins, seq leaves reporting a too-small Buffer to the shared + // write_no_io layer, which today makes this a silent truncation with exit 0. + test.concurrent("redirected to a Buffer smaller than the sequence", async () => { + const target = Buffer.alloc(100 * 1024); + const { stderr, exitCode } = await $`seq -s , -t END 1 ${COUNT} > ${target}`.nothrow().quiet(); + expect({ stderr: stderr.toString(), exitCode, target: target.toString() }).toEqual({ + stderr: "", + exitCode: 0, + target: expected.slice(0, target.length), + }); + }); + + // 8192 four-digit values with a 4-byte separator are exactly 64 KiB. Ending + // at 9191 the sequence runs out just as the first chunk fills up; ending at + // 9192 one value and the terminator are left over for a second chunk. + describe.each([ + [9191, 8192], + [9192, 8193], + ])("seq -s abcd -t END 1000 %i", (end, count) => { + const expected = Array.from({ length: count }, (_, i) => `${1000 + i}abcd`).join("") + "END"; + + test.concurrent("captured stdout", async () => { + expect(await $`seq -s abcd -t END 1000 ${end}`.text()).toBe(expected); + }); + + test.concurrent("redirected to a file", async () => { + using dir = tempDir("seq-boundary", {}); + const out = join(String(dir), "out.txt"); + await $`seq -s abcd -t END 1000 ${end} > ${out}`.quiet(); + expect(await Bun.file(out).text()).toBe(expected); + }); + + test.concurrent("piped to a process", async () => { + const stdout = await $`seq -s abcd -t END 1000 ${end} | ${bunExe()} -e ${copyStdin}`.env(bunEnv).text(); + expect(stdout).toBe(expected); + }); + }); + + // Once the reader is gone the chunks still to come fail with EPIPE and seq + // has to fail instead of hanging. A pipeline only reports the reader's exit + // code, so seq's own failure is made visible by chaining an `echo` to stderr + // off it; the drained reader shows the marker is not printed otherwise. + // 200k lines (~1.3 MB) are far more than a pipe buffers. + describe("fails once the reader is gone", () => { + const LINES = 200_000; + const run = async (pipeline: Promise<{ stdout: Buffer; stderr: Buffer; exitCode: number }>) => { + const { stdout, stderr, exitCode } = await pipeline; + return { stdout: stdout.toString(), stderr: stderr.toString(), exitCode }; + }; + + test.concurrent("reader exits after the first line", async () => { + const firstLine = + `const { value } = await Bun.stdin.stream().getReader().read();` + + `console.log(new TextDecoder().decode(value).split("\\n")[0]);` + + `process.exit(0);`; + const pipeline = $`(seq 1 ${LINES} || echo seq-failed 1>&2) | ${bunExe()} -e ${firstLine}`.env(bunEnv); + expect(await run(pipeline.nothrow().quiet())).toEqual({ stdout: "1\n", stderr: "seq-failed\n", exitCode: 0 }); + }); + + test.concurrent("reader never reads", async () => { + const pipeline = $`(seq 1 ${LINES} || echo seq-failed 1>&2) | true`; + expect(await run(pipeline.nothrow().quiet())).toEqual({ stdout: "", stderr: "seq-failed\n", exitCode: 0 }); + }); + + test.concurrent("reader drains the whole sequence", async () => { + const drain = "await Bun.stdin.bytes()"; + const pipeline = $`(seq 1 ${LINES} || echo seq-failed 1>&2) | ${bunExe()} -e ${drain}`.env(bunEnv); + expect(await run(pipeline.nothrow().quiet())).toEqual({ stdout: "", stderr: "", exitCode: 0 }); + }); + }); + + // seq used to render the whole sequence into one Vec before writing any of + // it, and IOWriter copies what it is handed, so writing N bytes to an fd + // took more than 2N bytes of memory: the child's RSS grew by about 90 MB for + // this ~30 MB sequence (130 MB under ASAN), and the freed buffers stay + // resident after the command (ASAN quarantines them, mimalloc keeps the + // pages). Streamed in chunks it grows by a couple of MB whatever the length. + // Measured in a child so nothing else in this file moves the numbers; the + // 100-byte separator makes the output large with few values, which keeps + // the child fast under ASAN. + test.concurrent("does not buffer the whole sequence before writing it", async () => { + await using proc = Bun.spawn({ + cmd: [ + bunExe(), + "-e", `const sep = Buffer.alloc(100, "x").toString();` + - `await Bun.$\`seq 1 10 > /dev/null\`;` + - `const b = rss();` + - `await Bun.$\`seq -s \${sep} 1 300000 > /dev/null\`;` + - `console.log(rss() - b);`, - ], - env: bunEnv, - stderr: "pipe", + `await Bun.$\`seq -s \${sep} 1 2000 > /dev/null\`;` + + `const before = process.memoryUsage.rss();` + + `await Bun.$\`seq -s \${sep} 1 300000 > /dev/null\`;` + + `console.log(process.memoryUsage.rss() - before);`, + ], + env: bunEnv, + stderr: "pipe", + }); + const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); + expect(stderr).toBe(""); + expect(stdout).toMatch(/^-?\d+\n$/); + expect(Number(stdout) / 1024 / 1024).toBeLessThan(16); + expect(exitCode).toBe(0); }); - const [stdout, stderr, exitCode] = await Promise.all([proc.stdout.text(), proc.stderr.text(), proc.exited]); - expect(stderr).toBe(""); - const rawDelta = stdout.trim(); - expect(rawDelta).toMatch(/^\d+$/); - const deltaBytes = Number(rawDelta); - // Output is 31_688_895 bytes. With the fix the child's RSS grows by - // ~128-134 MB (rendered Vec capacity + IOWriter's copy + ASAN shadow); - // without it the extra clone pushes it to ~170 MB. - expect(deltaBytes).toBeGreaterThan(0); - expect(deltaBytes).toBeLessThan(152 * 1024 * 1024); - expect(exitCode).toBe(0); });