From 9388c4a54abacf1a47f06c9eb7ac4d2a01029444 Mon Sep 17 00:00:00 2001 From: Mykhailo Chalyi Date: Wed, 7 Oct 2026 19:01:47 +0000 Subject: [PATCH] feat(builtins): stream grep and tr through pipelines grep and tr read their input pipe in record/UTF-8 aligned pieces and write results as they go, so `loop | grep y | head -2` and `yes | tr y n | head -1` end with SIGPIPE like bash. grep -m stops reading at its budget. Summarizing options stay buffered and drain the pipe (Context::stdin_to_end). Claude-Session: https://claude.ai/code/session_019aFikmptPc91Fj4N2iDXQA --- crates/bashkit/src/builtins/cat.rs | 5 +- crates/bashkit/src/builtins/cuttr.rs | 96 +++++++---- crates/bashkit/src/builtins/grep.rs | 156 +++++++++++++++++- crates/bashkit/src/builtins/mod.rs | 108 ++++++++++++ crates/bashkit/src/interpreter/mod.rs | 9 +- .../integration/limitations_evidence_tests.rs | 47 +++++- knowledge/foundations/parallel-execution.md | 8 + knowledge/log.md | 1 + knowledge/operations/limitations.md | 2 +- 9 files changed, 392 insertions(+), 40 deletions(-) diff --git a/crates/bashkit/src/builtins/cat.rs b/crates/bashkit/src/builtins/cat.rs index c459106ba..6b62b90fd 100644 --- a/crates/bashkit/src/builtins/cat.rs +++ b/crates/bashkit/src/builtins/cat.rs @@ -10,7 +10,7 @@ use async_trait::async_trait; use std::ffi::OsString; use std::path::Path; -use super::{Builtin, Context}; +use super::{Builtin, Context, STREAM_CHUNK_BYTES}; use crate::error::Result; use crate::fs::vfs_join; use crate::interpreter::ExecResult; @@ -121,9 +121,6 @@ impl Builtin for Cat { } } -/// Bytes per write when streaming into a pipeline (one pipe's capacity). -const STREAM_CHUNK_BYTES: usize = 4 * 1024; - /// 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). diff --git a/crates/bashkit/src/builtins/cuttr.rs b/crates/bashkit/src/builtins/cuttr.rs index b7c379ae4..d171e584e 100644 --- a/crates/bashkit/src/builtins/cuttr.rs +++ b/crates/bashkit/src/builtins/cuttr.rs @@ -344,22 +344,73 @@ impl Builtin for Tr { return Ok(Self::err("missing operand", 1)); } - let mut set1 = match expand_char_set(&non_flag_args[0]) { - Ok(set) => set, - Err(msg) => return Ok(Self::err(&msg, 1)), - }; + if let Err(msg) = expand_char_set(&non_flag_args[0]) { + return Ok(Self::err(&msg, 1)); + } if !delete && !squeeze && non_flag_args.len() < 2 { return Ok(Self::err("missing operand after SET1", 1)); } - let stdin = ctx.stdin.cloned().unwrap_or_default(); - let byte_mode = - ctx.env.get("LC_ALL").is_some_and(|locale| locale == "C") || stdin.text().is_err(); - if byte_mode && set1.iter().all(|c| (*c as u32) <= u8::MAX as u32) { - let set2 = if non_flag_args.len() >= 2 { - match expand_char_set(&non_flag_args[1]) { - Ok(set) => Some(set), + let spec = TrSpec { + args: &non_flag_args, + delete, + squeeze, + complement, + c_locale: ctx.env.get("LC_ALL").is_some_and(|locale| locale == "C"), + }; + + // A pipeline stage translates its input as it arrives, so + // `yes | tr y n | head -1` ends with SIGPIPE. Squeezing looks across + // piece boundaries, so `-s` reads everything first. + if !squeeze && let Some(out) = ctx.stdout_stream() { + let mut input = super::InputChunks::stdin(&ctx); + let cut = |data: &[u8]| { + if spec.c_locale { + data.len() + } else { + super::cut_utf8(data) + } + }; + while let Some(piece) = input.next(cut).await { + ctx.consume_budget_work(1)?; + let output = match spec.apply(&piece.into()) { + Ok(output) => output, Err(msg) => return Ok(Self::err(&msg, 1)), + }; + if !output.is_empty() && !out.write(&output).await { + return Ok(ExecResult::with_code(String::new(), 141)); } + } + return Ok(ExecResult::ok(String::new())); + } + + let stdin = ctx.stdin_to_end().await.unwrap_or_default(); + match spec.apply(&stdin) { + Ok(output) => Ok(ExecResult::ok_bytes(output)), + Err(msg) => Ok(Self::err(&msg, 1)), + } + } +} + +/// Parsed `tr` operands and flags. +struct TrSpec<'a> { + args: &'a [String], + delete: bool, + squeeze: bool, + complement: bool, + /// `LC_ALL=C`: work on bytes, not characters. + c_locale: bool, +} + +impl TrSpec<'_> { + /// Translate one piece of input; `Err` is a SET diagnostic. + fn apply(&self, stdin: &crate::StreamData) -> std::result::Result, String> { + let (non_flag_args, delete, squeeze, complement) = + (self.args, self.delete, self.squeeze, self.complement); + let mut set1 = expand_char_set(&non_flag_args[0])?; + let byte_mode = self.c_locale || stdin.text().is_err(); + if byte_mode && set1.iter().all(|c| (*c as u32) <= u8::MAX as u32) { + let set2 = if non_flag_args.len() >= 2 { + Some(expand_char_set(&non_flag_args[1])?) } else { None }; @@ -367,15 +418,14 @@ impl Builtin for Tr { .as_ref() .is_none_or(|set| set.iter().all(|c| (*c as u32) <= u8::MAX as u32)) { - let output = translate_bytes( + return Ok(translate_bytes( stdin.as_bytes(), &set1, set2.as_deref(), delete, squeeze, complement, - ); - return Ok(ExecResult::ok_bytes(output)); + )); } } if complement { @@ -389,15 +439,12 @@ impl Builtin for Tr { .collect(); } - let stdin = &*stdin; + let stdin: &str = stdin; let result = if delete && squeeze { // -ds: delete SET1 chars, then squeeze SET2 chars let set2 = if non_flag_args.len() >= 2 { - match expand_char_set(&non_flag_args[1]) { - Ok(set) => set, - Err(msg) => return Ok(Self::err(&msg, 1)), - } + expand_char_set(&non_flag_args[1])? } else { set1.clone() }; @@ -412,14 +459,7 @@ impl Builtin for Tr { // -s with only SET1: squeeze characters in SET1 squeeze_chars(stdin, &set1) } else { - if non_flag_args.len() < 2 { - return Ok(Self::err("missing operand after SET1", 1)); - } - - let set2 = match expand_char_set(&non_flag_args[1]) { - Ok(set) => set, - Err(msg) => return Ok(Self::err(&msg, 1)), - }; + let set2 = expand_char_set(&non_flag_args[1])?; let translated: String = stdin .chars() @@ -439,7 +479,7 @@ impl Builtin for Tr { } }; - Ok(ExecResult::ok(result)) + Ok(result.into_bytes()) } } diff --git a/crates/bashkit/src/builtins/grep.rs b/crates/bashkit/src/builtins/grep.rs index 86643fe24..911b8ede9 100644 --- a/crates/bashkit/src/builtins/grep.rs +++ b/crates/bashkit/src/builtins/grep.rs @@ -620,6 +620,23 @@ impl Builtin for Grep { let matcher = opts.build_matcher()?; + // A pipeline stage that prints matching lines streams them: lines + // go out as they are found, and `loop | grep y | head -1` ends with + // SIGPIPE. Options that summarize (-c, -l, -L, -q), look around (-A, + // -B, -C), offset (-b) or recurse keep the buffered path. + if !opts.recursive + && !opts.count_only + && !opts.files_with_matches + && !opts.files_without_matches + && !opts.quiet + && !opts.byte_offset + && opts.before_context == 0 + && opts.after_context == 0 + && let Some(out) = ctx.stdout_stream() + { + return stream_grep(&ctx, &opts, &matcher, &out).await; + } + let mut output = String::new(); // Diagnostics are a separate stream: a `grep: FILE: ...` line written // into `output` lands in the data a pipeline consumes, so @@ -650,7 +667,11 @@ impl Builtin for Grep { let inputs: Vec<(String, String)> = if opts.files.is_empty() { // Read from stdin // No NUL filtering here either — see `process_content`. - let stdin_content = ctx.stdin.map(ToString::to_string).unwrap_or_default(); + let stdin_content = ctx + .stdin_to_end() + .await + .map(|s| s.to_string()) + .unwrap_or_default(); vec![(stdin_name.to_string(), stdin_content)] } else if opts.recursive { // Try indexed search via SearchCapable if available. Skip it for -P: @@ -1075,6 +1096,139 @@ impl Builtin for Grep { } } +/// `grep` as a pipeline stage writing straight into the stage pipe: each +/// input is read in record-sized pieces and matching lines are written as +/// they are found. Same output as the buffered path for the options it +/// takes; a reader that went away ends it with 141 (SIGPIPE), and `-m` +/// stops reading as soon as the budget is spent. +async fn stream_grep( + ctx: &Context<'_>, + opts: &GrepOptions, + matcher: &Matcher, + out: &super::StdoutStream, +) -> Result { + let term = if opts.null_terminated { b'\0' } else { b'\n' }; + let fname_sep = if opts.null_filename { '\0' } else { ':' }; + let show_filename = !opts.no_filename && (opts.show_filename || opts.files.len() > 1); + let mut errors = String::new(); + let mut read_failed = false; + let mut any_match = false; + let sources: Vec> = if opts.files.is_empty() { + vec![None] + } else { + opts.files.iter().map(Some).collect() + }; + for source in sources { + let (name, mut input) = match source { + None => ("(standard input)", super::InputChunks::stdin(ctx)), + Some(file) => { + let path = if file.starts_with('/') { + std::path::PathBuf::from(file) + } else { + vfs_join(ctx.cwd, file) + }; + match ctx.fs.read_file(&path).await { + Ok(bytes) => (file.as_str(), super::InputChunks::bytes(bytes)), + Err(e) => { + read_failed = true; + if !opts.suppress_errors { + let reason = crate::error::io_error_reason(&e); + errors.push_str(&format!("grep: {file}: {reason}\n")); + } + continue; + } + } + } + }; + let mut line_no = 0usize; + let mut match_count = 0usize; + let mut binary = false; + let budget_spent = |count: usize| opts.max_count.is_some_and(|max| count >= max); + 'file: while let Some(piece) = input.next(super::cut_records(term)).await { + ctx.consume_budget_work(1)?; + // Binary input (a NUL without -a/-z): matches are reported once + // on stderr instead of printed; -I makes it a non-match. + if !binary && !opts.binary_as_text && !opts.null_terminated && piece.contains(&0) { + binary = true; + if opts.skip_binary { + break 'file; + } + } + let mut output = Vec::new(); + let mut stop = false; + for record in piece.split_inclusive(|&b| b == term) { + let record = record.strip_suffix(&[term]).unwrap_or(record); + let text = String::from_utf8_lossy(record); + // Same line text as `str::lines` in the buffered path. + let line = if term == b'\n' { + text.strip_suffix('\r').unwrap_or(&text) + } else { + &text + }; + line_no += 1; + let prefix = |output: &mut Vec| { + if show_filename { + output.extend_from_slice(name.as_bytes()); + output.push(fname_sep as u8); + } + if opts.line_numbers { + output.extend_from_slice(format!("{line_no}:").as_bytes()); + } + }; + if opts.only_matching && !opts.invert_match { + matcher.for_each_range(line, |(start, end)| { + if budget_spent(match_count) { + return false; + } + any_match = true; + match_count += 1; + if binary { + stop = true; + return false; + } + prefix(&mut output); + output.extend_from_slice(&line.as_bytes()[start..end]); + output.push(term); + true + }); + } else if matcher.is_match(line) != opts.invert_match { + any_match = true; + match_count += 1; + if binary { + stop = true; + } else { + prefix(&mut output); + output.extend_from_slice(line.as_bytes()); + output.push(term); + } + } + if stop || budget_spent(match_count) { + stop = true; + break; + } + } + if !output.is_empty() && !out.write(&output).await { + return Ok(ExecResult::err(errors, 141)); + } + if stop { + if binary && match_count > 0 { + // GNU grep >= 3.5: a diagnostic, never pipeline data. + errors.push_str(&format!("grep: {name}: binary file matches\n")); + } + break 'file; + } + } + } + let code = if read_failed { + 2 + } else if any_match { + 0 + } else { + 1 + }; + Ok(ExecResult::err(errors, code)) +} + /// Name a file found under a recursive-grep operand the way GNU grep does: /// the operand as typed, then the path below it (`d/` + `b` -> `d/b`, /// `.` + `b` -> `./b`). An empty operand is the implicit `-r` root and diff --git a/crates/bashkit/src/builtins/mod.rs b/crates/bashkit/src/builtins/mod.rs index 5f3d4814b..0ce94835f 100644 --- a/crates/bashkit/src/builtins/mod.rs +++ b/crates/bashkit/src/builtins/mod.rs @@ -968,6 +968,100 @@ impl StdinStream { } } +/// Bytes a streaming filter handles per step (one pipe's capacity). +pub(crate) const STREAM_CHUNK_BYTES: usize = 4 * 1024; + +/// Input of a streaming filter cut into pieces the filter can handle one at +/// a time: the data buffered before the stage started, then the stdin pipe +/// as it fills (or one file's bytes). Each piece ends where `cut` says (a +/// record end, a UTF-8 boundary) and is at most about +/// [`STREAM_CHUNK_BYTES`] unless one record is longer; the tail comes at +/// end of input. +pub(crate) struct InputChunks<'c> { + buf: Vec, + pos: usize, + buffered: Option<&'c [u8]>, + input: Option, + eof: bool, +} + +impl<'c> InputChunks<'c> { + /// Stdin of this builtin: `ctx.stdin`, then the streaming pipe if any. + pub(crate) fn stdin(ctx: &Context<'c>) -> Self { + Self { + buf: Vec::new(), + pos: 0, + buffered: ctx.stdin.map(|s| s.as_bytes()), + input: ctx.stdin_stream(), + eof: false, + } + } + + /// Input that is already all in memory (a file's contents). + pub(crate) fn bytes(bytes: Vec) -> Self { + Self { + buf: bytes, + pos: 0, + buffered: None, + input: None, + eof: true, + } + } + + /// Next piece, or `None` at end of input. `cut(data)` returns how many + /// leading bytes of `data` can be handled now (0 for "need more"). + pub(crate) async fn next(&mut self, cut: impl Fn(&[u8]) -> usize) -> Option> { + loop { + let rest = &self.buf[self.pos..]; + let window = rest.len().min(STREAM_CHUNK_BYTES); + let mut n = cut(&rest[..window]); + if n == 0 && rest.len() > window { + n = cut(rest); + } + if n == 0 && self.eof { + n = rest.len(); + } + if n > 0 { + let piece = rest[..n].to_vec(); + self.pos += n; + return Some(piece); + } + if self.eof { + return None; + } + self.buf.drain(..self.pos); + self.pos = 0; + if let Some(buffered) = self.buffered.take() { + self.buf.extend_from_slice(buffered); + continue; + } + let more = match &self.input { + Some(input) => input.read().await, + None => Vec::new(), + }; + if more.is_empty() { + self.eof = true; + } else { + self.buf.extend_from_slice(&more); + } + } + } +} + +/// [`InputChunks`] cut after the last `term`-terminated record. +pub(crate) fn cut_records(term: u8) -> impl Fn(&[u8]) -> usize { + move |data| data.iter().rposition(|&b| b == term).map_or(0, |p| p + 1) +} + +/// [`InputChunks`] cut at the last complete UTF-8 character; invalid bytes +/// are passed through so the filter can treat them as bytes. +pub(crate) fn cut_utf8(data: &[u8]) -> usize { + match std::str::from_utf8(data) { + Err(e) if e.error_len().is_none() => e.valid_up_to(), + _ => data.len(), + } +} + /// Stdout pipe of a streaming pipeline stage, see [`Context::stdout_stream`]. pub(crate) struct StdoutStream(std::sync::Arc); @@ -1082,6 +1176,20 @@ impl<'a> Context<'a> { .map(StdinStream) } + /// All of stdin, including what is still in a streaming stdin pipe, for + /// a filter stage that needs its whole input before it can write. + pub(crate) async fn stdin_to_end(&self) -> Option { + let Some(input) = self.stdin_stream() else { + return self.stdin.cloned(); + }; + let mut all = self + .stdin + .map(|s| s.as_bytes().to_vec()) + .unwrap_or_default(); + all.extend_from_slice(&input.read_to_end().await); + Some(all.into()) + } + /// 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 93e2f8d6f..ff4bba2d2 100644 --- a/crates/bashkit/src/interpreter/mod.rs +++ b/crates/bashkit/src/interpreter/mod.rs @@ -5862,14 +5862,17 @@ impl Interpreter { simple.redirects.is_empty() && matches!( simple.name.parts.as_slice(), - [WordPart::Literal(name)] if matches!(name.as_str(), "yes" | "seq" | "cat") + [WordPart::Literal(name)] + if matches!(name.as_str(), "yes" | "seq" | "cat" | "grep" | "tr") ) } /// Streaming stages that also read their stdin pipe incrementally - /// (`Context::stdin_stream`), so `loop | cat | head -1` stops early. + /// (`Context::stdin_stream`), so `loop | grep y | head -1` stops early. + /// They read the pipe to the end themselves when an option needs the + /// whole input (`grep -c`, `tr -s`). fn streams_stdin(name: &str) -> bool { - name == "cat" + matches!(name, "cat" | "grep" | "tr") } /// Run `commands` (the tail of a pipeline) concurrently: every stage but diff --git a/crates/bashkit/tests/integration/limitations_evidence_tests.rs b/crates/bashkit/tests/integration/limitations_evidence_tests.rs index b57d76381..13774a468 100644 --- a/crates/bashkit/tests/integration/limitations_evidence_tests.rs +++ b/crates/bashkit/tests/integration/limitations_evidence_tests.rs @@ -172,18 +172,41 @@ async fn l_term_002_cat_does_not_wait_for_terminal_input() { } /// L-PIPE-001: a leading single-builtin stage (other than the streaming -/// `yes`/`seq`/`cat`) runs to completion before `head` reads, and never -/// sees SIGPIPE: bash reports `141 0` here. +/// `yes`/`seq`/`cat`/`grep`/`tr`) 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; grep . /tmp/big | head -1; echo \"${PIPESTATUS[*]}\"") + .exec("seq 20000 > /tmp/big; sort -n /tmp/big | head -1; echo \"${PIPESTATUS[*]}\"") .await .unwrap(); assert_eq!(result.stdout, "1\n0 0\n"); } +/// Line filters stream too: `grep` and `tr` between an endless producer and +/// `head` stop with SIGPIPE, as in bash. +#[tokio::test] +async fn l_pipe_001_grep_tr_stream() { + let mut bash = Bash::new(); + let result = bash + .exec( + "seq 20000 > /tmp/big; grep . /tmp/big | head -1; echo \"${PIPESTATUS[*]}\"\n\ + while :; do echo y; done | grep y | head -2; echo \"${PIPESTATUS[*]}\"\n\ + yes | tr y n | head -1; echo \"${PIPESTATUS[*]}\"\n\ + yes | tr -d '\\n' | head -c 5; echo \" ${PIPESTATUS[*]}\"\n\ + yes | grep -n y | head -2\n\ + yes | grep -o -m3 y | cat; echo \"${PIPESTATUS[*]}\"", + ) + .await + .unwrap(); + assert_eq!( + result.stdout, + "1\n141 0\ny\ny\n141 141 0\nn\n141 141 0\nyyyyy 141 141 0\n1:y\n2:y\ny\ny\ny\n141 0 0\n" + ); + assert!(result.stderr.is_empty(), "{}", result.stderr); +} + /// 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] @@ -212,6 +235,24 @@ async fn l_pipe_001_generators_stream() { assert!(result.stderr.is_empty(), "{}", result.stderr); } +/// Options that need the whole input still see all of it when the stage +/// reads a streaming pipe (`grep -c`, `tr -s`, context lines). +#[tokio::test] +async fn l_pipe_001_buffered_filter_options_drain_pipe() { + let mut bash = Bash::new(); + let result = bash + .exec( + "src() { for i in 1 2 3 3 4; do echo \"a$i\"; done; }\n\ + src | grep -c a | cat\n\ + src | grep -A1 a2 | cat\n\ + src | tr -s 3 | tr -d '\\n' | cat; echo\n\ + src | grep -q a4 | cat; echo \"${PIPESTATUS[*]}\"", + ) + .await + .unwrap(); + assert_eq!(result.stdout, "5\na2\na3\na1a2a3a3a4\n0 0 0\n"); +} + /// L-PROC-004: child shells nest at most 8 deep. #[tokio::test] async fn l_proc_004_child_shell_depth() { diff --git a/knowledge/foundations/parallel-execution.md b/knowledge/foundations/parallel-execution.md index 4e20752b7..30f9e0c29 100644 --- a/knowledge/foundations/parallel-execution.md +++ b/knowledge/foundations/parallel-execution.md @@ -69,6 +69,14 @@ Implemented `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. +- `grep` and `tr` are streaming filters too. They pull input through + `InputChunks` (builtins/mod.rs): the data buffered before the stage + started, then the stdin pipe, cut into pieces of about 4 KiB that end on + a record (`grep`, `\n` or `\0` with `-z`) or a UTF-8 character (`tr`). + `grep` prints matches per piece and stops at `-m`; `tr` translates per + piece. Options that need the whole input stay buffered and drain the pipe + with `Context::stdin_to_end`: `grep -c/-l/-L/-q/-A/-B/-C/-b/-r`, + `tr -s` (a squeezed run could straddle two pieces). - 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 613d4cc64..e86d3ae89 100644 --- a/knowledge/log.md +++ b/knowledge/log.md @@ -6,6 +6,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 <