diff --git a/crates/bashkit/src/builtins/mod.rs b/crates/bashkit/src/builtins/mod.rs index 31b470e41..173de5457 100644 --- a/crates/bashkit/src/builtins/mod.rs +++ b/crates/bashkit/src/builtins/mod.rs @@ -950,6 +950,17 @@ pub struct Context<'a> { pub(crate) shell: Option>, } +/// Stdout pipe of a streaming pipeline stage, see [`Context::stdout_stream`]. +pub(crate) struct StdoutStream(std::sync::Arc); + +impl StdoutStream { + /// Wait for room in the pipe, then write. False once nobody reads. + pub(crate) async fn write(&self, data: &[u8]) -> bool { + self.0.writable().await; + self.0.write(data) + } +} + impl<'a> Context<'a> { /// Exact pipeline stdin bytes. pub fn stdin_bytes(&self) -> Option<&[u8]> { @@ -1032,6 +1043,17 @@ impl<'a> Context<'a> { .and_then(|shell| shell.execution_extensions.get::()) } + /// Streaming stdout when this builtin is a producer stage of a + /// concurrent pipeline (`yes | head -1`). Write chunks as they are made + /// instead of returning them in `ExecResult::stdout`; when a write + /// reports the reader gone, stop and exit 141 (SIGPIPE). + pub(crate) fn stdout_stream(&self) -> Option { + self.shell + .as_ref() + .and_then(|shell| shell.stdout_pipe.clone()) + .map(StdoutStream) + } + /// 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/builtins/seq.rs b/crates/bashkit/src/builtins/seq.rs index 607392612..4daf0036e 100644 --- a/crates/bashkit/src/builtins/seq.rs +++ b/crates/bashkit/src/builtins/seq.rs @@ -18,6 +18,9 @@ use crate::interpreter::ExecResult; /// -w Equalize width by padding with leading zeroes pub struct Seq; +/// Bytes per write when streaming into a pipeline (one pipe's capacity). +const STREAM_CHUNK_BYTES: usize = 4 * 1024; + #[async_trait] impl Builtin for Seq { async fn execute(&self, ctx: Context<'_>) -> Result { @@ -140,6 +143,10 @@ impl Builtin for Seq { }; let mut output = String::new(); + // Pipeline producer: flush full chunks as they are made, so + // `seq 1000000 | head -1` stops early with SIGPIPE. + let stream = ctx.stdout_stream(); + let mut flushed = 0usize; let mut current = first; let mut first_item = true; @@ -160,7 +167,17 @@ impl Builtin for Seq { capped = Some(("line", SEQ_MAX_LINES)); break; } - if output.len() > SEQ_MAX_OUTPUT_BYTES { + if let Some(stream) = &stream + && output.len() >= STREAM_CHUNK_BYTES + { + ctx.consume_budget_work(1)?; + if !stream.write(output.as_bytes()).await { + return Ok(ExecResult::with_code("", 141)); + } + flushed += output.len(); + output.clear(); + } + if flushed + output.len() > SEQ_MAX_OUTPUT_BYTES { capped = Some(("output byte", SEQ_MAX_OUTPUT_BYTES)); break; } @@ -187,9 +204,15 @@ impl Builtin for Seq { current += increment; } - if !output.is_empty() { + if !first_item { output.push('\n'); } + if let Some(stream) = &stream { + if !stream.write(output.as_bytes()).await { + return Ok(ExecResult::with_code("", 141)); + } + output.clear(); + } if let Some((what, limit)) = capped { return Ok(cap_exceeded("seq", output, what, limit)); diff --git a/crates/bashkit/src/builtins/yes.rs b/crates/bashkit/src/builtins/yes.rs index f00dd3faa..483e3ebdb 100644 --- a/crates/bashkit/src/builtins/yes.rs +++ b/crates/bashkit/src/builtins/yes.rs @@ -15,6 +15,9 @@ use crate::interpreter::ExecResult; /// In bashkit, output is limited to avoid infinite loops. pub struct Yes; +/// Bytes per write when streaming into a pipeline (one pipe's capacity). +const STREAM_CHUNK_BYTES: usize = 4 * 1024; + impl BuiltinHelper for Yes { const NAME: &'static str = "yes"; } @@ -66,6 +69,23 @@ impl Builtin for Yes { // THREAT[TM-DOS-109]: the cap always ends `yes`; say so on stderr so a // consumer wanting more lines than the cap can tell output was cut. let output = build_yes_output(&text); + if let Some(stream) = ctx.stdout_stream() { + // Pipeline producer: hand the reader one chunk at a time, so + // `yes | head -1` stops after the first chunk with SIGPIPE. + let lines = output.lines().count(); + for chunk in output.as_bytes().chunks(STREAM_CHUNK_BYTES) { + ctx.consume_budget_work(1)?; + if !stream.write(chunk).await { + return Ok(ExecResult::with_code("", 141)); + } + } + return Ok(super::limits::cap_exceeded( + "yes", + String::new(), + "output", + format!("{lines} lines"), + )); + } let lines = output.lines().count(); Ok(super::limits::cap_exceeded( "yes", diff --git a/crates/bashkit/src/interpreter/mod.rs b/crates/bashkit/src/interpreter/mod.rs index 279ee296a..e755265ed 100644 --- a/crates/bashkit/src/interpreter/mod.rs +++ b/crates/bashkit/src/interpreter/mod.rs @@ -16,7 +16,7 @@ mod brace_expansion; mod expansion; mod glob; mod jobs; -mod pipe; +pub(crate) mod pipe; mod redirection; mod state; mod time_command; @@ -366,6 +366,8 @@ pub(crate) struct ShellRef<'a> { pub(crate) jobs: &'a SharedJobTable, /// Typed per-execution extensions for the current `exec*()` call. pub(crate) execution_extensions: Arc, + /// Stdout of a streaming pipeline stage (see `Context::stdout_stream`). + pub(crate) stdout_pipe: Option>, } // Interpreter-dispatched "special" builtins, listed here so the public @@ -1413,6 +1415,12 @@ pub struct Interpreter { /// Stdout pipe of this forked pipeline stage: command boundaries wait /// here for room and stop with 141 once the reader is gone. pipe_out: Option>, + /// Address of the stage's own simple command when it is a streaming + /// builtin (`yes`, `seq`): only that command writes straight into + /// `pipe_out`, never a command run from its argument expansions. + stream_stdout_command: Option, + /// Pipe handed to the next builtin's `Context` (taken at dispatch). + builtin_stdout_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". @@ -1986,6 +1994,8 @@ impl Interpreter { pipeline_stdin: None, pipe_in: None, pipe_out: None, + stream_stdout_command: None, + builtin_stdout_pipe: None, getopts_char_idx: 0, last_bg_pid: None, output_callback: None, @@ -2141,6 +2151,8 @@ impl Interpreter { pipeline_stdin: None, pipe_in: None, pipe_out: None, + stream_stdout_command: None, + builtin_stdout_pipe: None, getopts_char_idx: self.getopts_char_idx, last_bg_pid: self.last_bg_pid.clone(), output_callback: None, @@ -5826,6 +5838,19 @@ impl Interpreter { !matches!(name, "eval" | "source" | "." | "bash" | "sh") && !self.scoped.functions.contains_key(name) && !self.scoped.aliases.contains_key(name) + && !Self::streams_stdout(simple) + } + + /// Builtins that write their stdout to a pipeline as they go + /// (`Context::stdout_stream`): generators whose output is unbounded or + /// large, so `yes | head -1` and `seq 1000000 | head -1` stop at the + /// first line with SIGPIPE instead of running into the output caps. + fn streams_stdout(simple: &SimpleCommand) -> bool { + simple.redirects.is_empty() + && matches!( + simple.name.parts.as_slice(), + [WordPart::Literal(name)] if matches!(name.as_str(), "yes" | "seq") + ) } /// Run `commands` (the tail of a pipeline) concurrently: every stage but @@ -5867,6 +5892,11 @@ impl Interpreter { let command = command.clone(); stages.push(Box::pin(async move { let _read_end = read_end; + if let Command::Simple(simple) = &command + && Self::streams_stdout(simple) + { + child.stream_stdout_command = Some(simple as *const SimpleCommand as usize); + } let jobs = Arc::clone(&child.jobs); let result = jobs::with_jobs(&jobs, child.execute_command(&command)).await; jobs.finish_all().await; @@ -6951,6 +6981,16 @@ impl Interpreter { stdin: Option, ) -> std::pin::Pin> + Send + 'a>> { Box::pin(async move { + // The stage's own `yes`/`seq` writes straight into the pipe. + self.builtin_stdout_pipe = if self.stream_stdout_command + == Some(command as *const SimpleCommand as usize) + && command.redirects.is_empty() + && !self.scoped.functions.contains_key(name) + { + self.pipe_out.clone() + } 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()); @@ -7354,6 +7394,7 @@ impl Interpreter { limits: &self.limits, jobs: &self.jobs, execution_extensions, + stdout_pipe: None, }; let plan_ctx = builtins::Context { args, @@ -7419,6 +7460,7 @@ impl Interpreter { limits: &self.limits, jobs: &self.jobs, execution_extensions, + stdout_pipe: self.builtin_stdout_pipe.take(), }; let ctx = builtins::Context { args, diff --git a/crates/bashkit/tests/integration/limitations_evidence_tests.rs b/crates/bashkit/tests/integration/limitations_evidence_tests.rs index 8bb2304c7..f566bfafb 100644 --- a/crates/bashkit/tests/integration/limitations_evidence_tests.rs +++ b/crates/bashkit/tests/integration/limitations_evidence_tests.rs @@ -171,17 +171,29 @@ async fn l_term_002_cat_does_not_wait_for_terminal_input() { ); } -/// L-PIPE-001: a leading single-builtin stage runs to its own cap before -/// `head` reads, and never sees SIGPIPE. +/// L-PIPE-001: a leading single-builtin stage (other than the streaming +/// generators `yes`/`seq`) 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[*]}\"") + .await + .unwrap(); + assert_eq!(result.stdout, "1\n0 0\n"); +} + +/// The generators stream: `yes | head -1` stops with SIGPIPE, as in bash. +#[tokio::test] +async fn l_pipe_001_generators_stream() { let mut bash = Bash::new(); let result = bash .exec("yes | head -1; echo \"${PIPESTATUS[*]}\"") .await .unwrap(); - assert_eq!(result.stdout, "y\n1 0\n"); - assert!(result.stderr.contains("output limit"), "{}", result.stderr); + assert_eq!(result.stdout, "y\n141 0\n"); + assert!(result.stderr.is_empty(), "{}", result.stderr); } /// L-PROC-004: child shells nest at most 8 deep. diff --git a/crates/bashkit/tests/spec_cases/bash/pipe-generators.test.sh b/crates/bashkit/tests/spec_cases/bash/pipe-generators.test.sh new file mode 100644 index 000000000..e90543312 --- /dev/null +++ b/crates/bashkit/tests/spec_cases/bash/pipe-generators.test.sh @@ -0,0 +1,38 @@ +# Generators at the head of a pipeline stop when the reader goes away +# (SIGPIPE, status 141) instead of running to an output cap. + +### yes_head_sigpipe +yes | head -2 +echo "${PIPESTATUS[*]}" +### expect +y +y +141 0 +### end + +### seq_million_head +seq 1000000 | head -1 +echo "${PIPESTATUS[*]}" +### expect +1 +141 0 +### end + +### seq_full_output_through_pipe +seq 5000 | tail -1 +seq -s, 5 | cat +seq 3 | wc -l +### expect +5000 +1,2,3,4,5 +3 +### end + +### yes_custom_text_head_bytes +yes ab | head -c 7 +echo +### expect +ab +ab +a +### end diff --git a/knowledge/foundations/parallel-execution.md b/knowledge/foundations/parallel-execution.md index ed0bd0898..82b959429 100644 --- a/knowledge/foundations/parallel-execution.md +++ b/knowledge/foundations/parallel-execution.md @@ -57,6 +57,13 @@ Implemented first, one after another, on the cheap subshell snapshot (`enter_pipeline_stage`/`exit_pipeline_stage`): a builtin produces its output as one value anyway, and this keeps `echo x | grep x` fork-free. +- The generators `yes` and `seq` (unredirected, not shadowed by a + function) also start the concurrent part: as a producer stage they write + 4 KiB chunks straight into the pipe through `Context::stdout_stream` + (the interpreter hands the pipe only to the stage's own simple command, + 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. - 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 196d2b0cc..d0f88b8bf 100644 --- a/knowledge/log.md +++ b/knowledge/log.md @@ -2,6 +2,7 @@ ## 2026-10-07 +* **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 <