diff --git a/crates/bashkit/src/builtins/cat.rs b/crates/bashkit/src/builtins/cat.rs index 620feb18c..c459106ba 100644 --- a/crates/bashkit/src/builtins/cat.rs +++ b/crates/bashkit/src/builtins/cat.rs @@ -63,12 +63,26 @@ impl Builtin for Cat { .map(|vs| vs.map(|v| v.to_string_lossy().into_owned()).collect()) .unwrap_or_default(); + let plain = !(show_ends + || show_tabs + || show_nonprinting + || number_all + || number_nonblank + || squeeze); + if plain && let Some(stream) = ctx.stdout_stream() { + return stream_plain(&ctx, &files, &stream).await; + } + let mut raw = Vec::new(); + let mut stderr = String::new(); for file in &files { if file == "-" { if let Some(stdin) = ctx.stdin { raw.extend_from_slice(stdin.as_bytes()); } + if let Some(input) = ctx.stdin_stream() { + raw.extend_from_slice(&input.read_to_end().await); + } } else { let path = if Path::new(file).is_absolute() { file.clone() @@ -77,30 +91,95 @@ impl Builtin for Cat { }; match ctx.fs.read_file(Path::new(&path)).await { Ok(bytes) => raw.extend_from_slice(&bytes), + // GNU cat reports the operand and goes on with the rest. Err(e) => { let reason = crate::error::io_error_reason(&e); - return Ok(ExecResult::err(format!("cat: {file}: {reason}\n"), 1)); + stderr.push_str(&format!("cat: {file}: {reason}\n")); } } } } - if !(show_ends || show_tabs || show_nonprinting || number_all || number_nonblank || squeeze) - { - return Ok(ExecResult::ok_bytes(raw)); + let mut result = if plain { + ExecResult::ok_bytes(raw) + } else { + ExecResult::ok_bytes(render( + &raw, + show_ends, + show_tabs, + show_nonprinting, + number_all, + number_nonblank, + squeeze, + )) + }; + if !stderr.is_empty() { + result.stderr = stderr.into(); + result.exit_code = 1; } + Ok(result) + } +} + +/// Bytes per write when streaming into a pipeline (one pipe's capacity). +const STREAM_CHUNK_BYTES: usize = 4 * 1024; - let output = render( - &raw, - show_ends, - show_tabs, - show_nonprinting, - number_all, - number_nonblank, - squeeze, - ); - Ok(ExecResult::ok(output)) +/// Plain `cat` as a pipeline stage: copy each operand into the stage pipe +/// as it is read, stdin chunk by chunk, so `loop | cat | head -1` and +/// `cat big | head -1` stop early. A closed reader ends with 141 (SIGPIPE). +async fn stream_plain( + ctx: &Context<'_>, + files: &[String], + stream: &super::StdoutStream, +) -> Result { + async fn send(ctx: &Context<'_>, stream: &super::StdoutStream, data: &[u8]) -> Result { + for chunk in data.chunks(STREAM_CHUNK_BYTES) { + ctx.consume_budget_work(1)?; + if !stream.write(chunk).await { + return Ok(false); + } + } + Ok(true) + } + let mut stderr = String::new(); + let mut code = 0; + for file in files { + let sent = if file == "-" { + let mut ok = match ctx.stdin { + Some(stdin) => send(ctx, stream, stdin.as_bytes()).await?, + None => true, + }; + if let Some(input) = ctx.stdin_stream() { + while ok { + let chunk = input.read().await; + if chunk.is_empty() { + break; + } + ok = send(ctx, stream, &chunk).await?; + } + } + ok + } else { + let path = if Path::new(file).is_absolute() { + file.clone() + } else { + vfs_join(ctx.cwd, file).to_string_lossy().into_owned() + }; + match ctx.fs.read_file(Path::new(&path)).await { + Ok(bytes) => send(ctx, stream, &bytes).await?, + Err(e) => { + let reason = crate::error::io_error_reason(&e); + stderr.push_str(&format!("cat: {file}: {reason}\n")); + code = 1; + true + } + } + }; + if !sent { + return Ok(ExecResult::err(stderr, 141)); + } } + Ok(ExecResult::err(stderr, code)) } /// Apply cat's display transforms in a single pass. diff --git a/crates/bashkit/src/builtins/mod.rs b/crates/bashkit/src/builtins/mod.rs index 7cadee453..4765dccbc 100644 --- a/crates/bashkit/src/builtins/mod.rs +++ b/crates/bashkit/src/builtins/mod.rs @@ -952,6 +952,21 @@ pub struct Context<'a> { pub(crate) shell: Option>, } +/// Stdin pipe of a streaming filter stage, see [`Context::stdin_stream`]. +pub(crate) struct StdinStream(std::sync::Arc); + +impl StdinStream { + /// Next chunk of input; empty at end of input. + pub(crate) async fn read(&self) -> Vec { + self.0.read_some().await + } + + /// Everything left, for filters that need their whole input. + pub(crate) async fn read_to_end(&self) -> Vec { + self.0.read_to_end().await + } +} + /// Stdout pipe of a streaming pipeline stage, see [`Context::stdout_stream`]. pub(crate) struct StdoutStream(std::sync::Arc); @@ -1056,6 +1071,16 @@ impl<'a> Context<'a> { .map(StdoutStream) } + /// Streaming stdin when this builtin is a filter stage of a concurrent + /// pipeline (`loop | cat | head -1`). `ctx.stdin` then holds only data + /// already buffered before the stage started; read the rest here. + pub(crate) fn stdin_stream(&self) -> Option { + self.shell + .as_ref() + .and_then(|shell| shell.stdin_pipe.clone()) + .map(StdinStream) + } + /// Remaining wall-clock budget of the current `exec*` call, if limited. pub fn remaining_deadline(&self) -> Option { self.execution_extension::()? diff --git a/crates/bashkit/src/interpreter/mod.rs b/crates/bashkit/src/interpreter/mod.rs index 633c6bbd5..7582fb6a6 100644 --- a/crates/bashkit/src/interpreter/mod.rs +++ b/crates/bashkit/src/interpreter/mod.rs @@ -368,6 +368,8 @@ pub(crate) struct ShellRef<'a> { pub(crate) execution_extensions: Arc, /// Stdout of a streaming pipeline stage (see `Context::stdout_stream`). pub(crate) stdout_pipe: Option>, + /// Stdin of a streaming filter stage (see `Context::stdin_stream`). + pub(crate) stdin_pipe: Option>, } // Interpreter-dispatched "special" builtins, listed here so the public @@ -1421,6 +1423,8 @@ pub struct Interpreter { stream_stdout_command: Option, /// Pipe handed to the next builtin's `Context` (taken at dispatch). builtin_stdout_pipe: Option>, + /// Input pipe handed to a streaming filter's `Context` (taken at dispatch). + builtin_stdin_pipe: Option>, /// Position within the current argument while `getopts` walks a clustered /// short-option group (e.g. `-abc`). Interpreter-internal working state for /// `execute_getopts`; `0` means "at the start of the next option group". @@ -1997,6 +2001,7 @@ impl Interpreter { pipe_out: None, stream_stdout_command: None, builtin_stdout_pipe: None, + builtin_stdin_pipe: None, getopts_char_idx: 0, last_bg_pid: None, output_callback: None, @@ -2154,6 +2159,7 @@ impl Interpreter { pipe_out: None, stream_stdout_command: None, builtin_stdout_pipe: None, + builtin_stdin_pipe: None, getopts_char_idx: self.getopts_char_idx, last_bg_pid: self.last_bg_pid.clone(), output_callback: None, @@ -5856,10 +5862,16 @@ impl Interpreter { simple.redirects.is_empty() && matches!( simple.name.parts.as_slice(), - [WordPart::Literal(name)] if matches!(name.as_str(), "yes" | "seq") + [WordPart::Literal(name)] if matches!(name.as_str(), "yes" | "seq" | "cat") ) } + /// Streaming stages that also read their stdin pipe incrementally + /// (`Context::stdin_stream`), so `loop | cat | head -1` stops early. + fn streams_stdin(name: &str) -> bool { + name == "cat" + } + /// Run `commands` (the tail of a pipeline) concurrently: every stage but /// the last on a forked shell writing into a bounded [`pipe::Pipe`], the /// last one here (a subshell unless `lastpipe`) so its output streams. @@ -7001,6 +7013,14 @@ impl Interpreter { } else { None }; + // A streaming filter (`cat`) reads its input pipe as it goes + // instead of having it collected up front. + self.builtin_stdin_pipe = + if self.builtin_stdout_pipe.is_some() && Self::streams_stdin(name) { + self.pipe_in.take() + } else { + None + }; // Track $_ (last argument of previous command, from already-expanded args) if let Some(last) = args.last() { self.insert_variable_checked("_".to_string(), last.clone()); @@ -7405,6 +7425,7 @@ impl Interpreter { jobs: &self.jobs, execution_extensions, stdout_pipe: None, + stdin_pipe: None, }; let plan_ctx = builtins::Context { args, @@ -7471,6 +7492,7 @@ impl Interpreter { jobs: &self.jobs, execution_extensions, stdout_pipe: self.builtin_stdout_pipe.take(), + stdin_pipe: self.builtin_stdin_pipe.take(), }; let ctx = builtins::Context { args, diff --git a/crates/bashkit/src/interpreter/pipe.rs b/crates/bashkit/src/interpreter/pipe.rs index e03dd47d4..14f404c05 100644 --- a/crates/bashkit/src/interpreter/pipe.rs +++ b/crates/bashkit/src/interpreter/pipe.rs @@ -164,7 +164,6 @@ impl Pipe { } /// Read everything until the writer closes. - #[cfg(test)] pub(crate) async fn read_to_end(&self) -> Vec { let mut out = Vec::new(); loop { diff --git a/crates/bashkit/tests/integration/limitations_evidence_tests.rs b/crates/bashkit/tests/integration/limitations_evidence_tests.rs index fbedaf3e4..b57d76381 100644 --- a/crates/bashkit/tests/integration/limitations_evidence_tests.rs +++ b/crates/bashkit/tests/integration/limitations_evidence_tests.rs @@ -172,18 +172,34 @@ async fn l_term_002_cat_does_not_wait_for_terminal_input() { } /// L-PIPE-001: a leading single-builtin stage (other than the streaming -/// generators `yes`/`seq`) runs to completion before `head` reads, and never +/// `yes`/`seq`/`cat`) runs to completion before `head` reads, and never /// sees SIGPIPE: bash reports `141 0` here. #[tokio::test] async fn l_pipe_001_stages_run_sequentially() { let mut bash = Bash::new(); let result = bash - .exec("seq 20000 > /tmp/big; cat /tmp/big | head -1; echo \"${PIPESTATUS[*]}\"") + .exec("seq 20000 > /tmp/big; grep . /tmp/big | head -1; echo \"${PIPESTATUS[*]}\"") .await .unwrap(); assert_eq!(result.stdout, "1\n0 0\n"); } +/// Plain `cat` streams as a stage, so it gets SIGPIPE like bash, and a +/// `cat` between an endless loop and `head` no longer runs to a limit. +#[tokio::test] +async fn l_pipe_001_cat_streams() { + let mut bash = Bash::new(); + let result = bash + .exec( + "seq 20000 > /tmp/big; cat /tmp/big | head -1; echo \"${PIPESTATUS[*]}\"\n\ + while :; do echo y; done | cat | head -2; echo \"${PIPESTATUS[*]}\"", + ) + .await + .unwrap(); + assert_eq!(result.stdout, "1\n141 0\ny\ny\n141 141 0\n"); + assert!(result.stderr.is_empty(), "{}", result.stderr); +} + /// The generators stream: `yes | head -1` stops with SIGPIPE, as in bash. #[tokio::test] async fn l_pipe_001_generators_stream() { diff --git a/knowledge/foundations/parallel-execution.md b/knowledge/foundations/parallel-execution.md index 82b959429..4e20752b7 100644 --- a/knowledge/foundations/parallel-execution.md +++ b/knowledge/foundations/parallel-execution.md @@ -64,6 +64,11 @@ Implemented matched by address, never to a command run from its arguments), so `seq 1000000 | head -1` stops after one chunk with `PIPESTATUS` `141 0`. Their output caps still apply when the reader takes everything. +- Plain `cat` (no display flags) streams the same way and is also a + streaming filter: the interpreter hands it its input pipe through + `Context::stdin_stream` instead of collecting stdin up front, so + `while :; do echo; done | cat | head -1` stops with SIGPIPE. `cat` with + flags drains the input pipe and runs buffered. - From the first stage that runs shell code (loop, group, function, `eval`, nested shell), the rest run concurrently (`execute_streaming_stages`, on with `concurrent_jobs`). Each non-last stage is a forked interpreter like a diff --git a/knowledge/log.md b/knowledge/log.md index 09dd188a0..a834904b1 100644 --- a/knowledge/log.md +++ b/knowledge/log.md @@ -4,6 +4,7 @@ * **Feature**: `yes` and `seq` stream into pipelines. As a producer stage they write chunks into the pipe as they go (`Context::stdout_stream`), so `yes | head -1` and `seq 1000000 | head -1` stop after the first chunk with `PIPESTATUS` `141 0` instead of running to the output cap and printing a truncation notice. Step 2 of the roadmap (streaming builtin I/O), starting with the generators. * **Feature**: `sudo` and `busybox`, the top missing commands in the gap replay (`sudo tee /etc/x <