Feat/thread pool processing

Changelog.md

@@ -1,6 +1,16 @@
 # Changelog
 
 - 0.6.0
+    - `each --threads=N` runs up to N commands in parallel, as a threadpool. Each command's output
+      is buffered and written in one piece, so parallel commands can't mangle each other's output.
+      Results are printed in the order the commands finish; `each --sequential` prints them in the
+      order of the input instead, waiting for earlier lines.
+    - `each` draws a progress bar on stderr while more than one thread is running, leaving stdout
+      free for piping
+    - **breaking**: a command failing under `each` no longer aborts the run. The failure is reported
+      on stderr, the remaining lines still run, and `string` exits 1 at the end.
+    - errors are now printed instead of being swallowed by a silent `exit(1)`, and the output of a
+      failed command is actually included in the message (it was always empty before)
     - **breaking**: `contains`, `starts-with` and `ends-with` are no longer predicates over the whole
       input. Instead of exiting 0/1 they now filter the input line by line and print every matching
       line. Shell conditions relying on the exit code (`string contains foo && ...`) need to be

README.md

@@ -12,21 +12,24 @@ Cli for common string operations. Takes input from stdin.
 Usage: string <COMMAND>
 
 Commands:
-  case        Transform upper- or lowercase
-  reverse     Reverse order of lines
-  substr      Extract part of a given string
-  split       Split up a string by a separator and print the parts on separate lines
-  length      Returns the length the input string
-  replace     Replace all matching characters
-  line        Pick a single line by index
-  interleave  Interleave input and only print every nth line
-  distinct    Output the set of input strings without repetitions, in order
-  trim        Trim whitespace on lines and ignore empty ones
-  chars       Prints all chars on separate lines
-  template    Useful for templating, replace sections of input with the output of a shell command or script
-  Map each line of input to a subcommand.
-  each        
-  help        Print this message or the help of the given subcommand(s)
+  case         Transform upper- or lowercase
+  reverse      Reverse order of lines
+  substr       Extract part of a given string
+  split        Split up a string by a separator and print the parts on separate lines
+  join         Join lines with a separator into a single string
+  contains     Print all lines containing the given string
+  starts-with  Print all lines starting with the given prefix, ignoring leading whitespace
+  ends-with    Print all lines ending with the given suffix, ignoring trailing whitespace
+  length       Returns the length the input string
+  replace      Replace all matching characters
+  line         Pick a single line by index
+  interleave   Interleave input and only print every nth line
+  distinct     Output the set of input strings without repetitions, in order
+  trim         Trim whitespace on lines and ignore empty ones
+  chars        Prints all chars on separate lines
+  template     Useful for templating, replace sections of input with the output of a shell command or script
+  each         Map each line of input to a subcommand
+  help         Print this message or the help of the given subcommand(s)
 
 Options:
   -h, --help  Print help
@@ -40,9 +43,11 @@ This is mostly because there are thousands of ways to do the tasks `shell-string
 More than anything I hated finding some solution for file templating over and over again. I wrote `shell-string` to never again have to think about what the best way of templating a file is.
 It's always this, period.
 
-## Why is `shell-string` good for templating files?
+## Template Files
+`shell-string` is good for templating files.
 
-Because you practically have no restrictions.
+It's a very simple and clean solution where
+you practically have no restrictions.
 You need to just drop in some environment variables? Easy, just write `{{ echo $MY_VAR }}` into the template.
 Is complex logic needed? You could write `{{ console.log(crazyStuff()) }}` and you're golden. Just execute with `--shell=node`.
 You want to use `haskell` in your template files? Use `--shell=ghci`!
@@ -81,6 +86,65 @@ which means
 - `| string template`:              The `|` means "don't print this in a terminal, pipe it to another programm" and that programm is `string` in `template` mode.
 - `> deployment.yaml`:              Write the output of this into a file called `deployment.yaml`. If the file existed, empty it beforehand.
 
+## Use It As A Threadpool
+
+`string each` runs a command once per line of input. Waiting for those commands one after another
+is pure waste whenever they sit on the network instead of the CPU, so `--threads` turns `each` into
+a threadpool:
+
+```sh
+cat urls.txt | string each --threads=12 -- curl -s {}
+```
+
+Twelve requests are in flight at any moment, and `string` keeps the pool full: whenever a command
+finishes, the next line is handed to the thread that just became free. Nothing is scheduled up
+front, so one slow url doesn't leave eleven threads idling.
+
+The thing that makes this usable rather than a mess is that **output is never interleaved**. The
+output of a command is collected in full and written in one piece, so twelve `curl`s can't scribble
+over each other halfway through a line. What you get is the same output you'd get from a sequential
+run, just sooner.
+
+### Watching it work
+
+Because output is buffered, a long run would otherwise look like a hung terminal. So while more than
+one thread is running, a progress bar is drawn — on stderr, never stdout:
+
+```
+[####################----------] 812/1200 (67%)
+```
+
+That split is deliberate: `... | string each -t 12 -- curl -s {} > results.txt` shows you the bar in
+the terminal while `results.txt` receives nothing but results.
+
+### Ordering
+
+Results appear in the order the commands _finish_, which is what you want when you're watching them
+come in. When you'd rather have them line up with your input, ask for it:
+
+```sh
+cat urls.txt | string each --threads=12 --sequential -- curl -s {}
+```
+
+`--sequential` holds finished results back until every earlier line is done, so the output matches
+the input line for line. The work still runs on all twelve threads — only the printing waits.
+
+### When something fails
+
+One broken url shouldn't throw away the other 1199 responses. A command that exits non-zero is
+reported on stderr, naming the line it came from, and the run carries on:
+
+```
+line 47: error executing command `curl`.
+Process terminated with exit code exit status: 6.
+Program output:
+curl: (6) Could not resolve host: exmaple.com
+```
+
+At the very end `string` prints how many commands failed and exits 1, so `&&` in a script still
+does the right thing — you just get all the successful output, and a full list of what went wrong,
+instead of everything stopping at the first problem.
+
 ## Installation
 
 Given cargo is installed on your machine execute

src/cli.rs

@@ -118,6 +118,13 @@ pub enum StringCommand {
         /// Name of placeholder in command to be replaced with input. Default is "{}"
         #[arg(short = 'v', long = "var", default_value = "{}")]
         var: String,
+        /// Number of commands to run in parallel. The default of 1 runs them one after another
+        #[arg(short = 't', long = "threads", default_value_t = 1)]
+        threads: usize,
+        /// Print the results in the order of the input, instead of the order they finish in.
+        /// Waits for earlier lines, so a single slow command holds back everything behind it
+        #[arg(long = "sequential", default_value_t = false)]
+        sequential: bool,
         /// Command to be executed. Pass "string each -- <commands...>" so you can pass flags to the command.
         command: Vec<String>,
     },

src/exec.rs

@@ -1,39 +1,63 @@
+use anyhow::{bail, Context};
 use std::{io::Write, process::Stdio};
 
-pub fn execute(command: &[String], stdin_text: Option<&str>) -> String {
+pub fn execute(command: &[String], stdin_text: Option<&str>) -> anyhow::Result<String> {
     let command_name = &command[0];
-    let mut command = std::process::Command::new(&command[0])
+    let mut child = std::process::Command::new(&command[0])
         .args(&command[1..])
         .stdin(Stdio::piped())
         .stdout(Stdio::piped())
+        // captured, so the output of parallel commands can't smear over each other or the
+        // progress bar. It is printed as part of the error message if the command fails.
+        .stderr(Stdio::piped())
         .spawn()
-        .expect(&format!("failed to spawn process {:?}", command));
-
-    if let Some(stdin_text) = stdin_text {
-        let stdin = command
-            .stdin
-            .as_mut()
-            .expect("failed to open stdin of command");
-        stdin
-            .write_all(stdin_text.as_bytes())
-            .expect("failed to pipe command into shell");
-    }
+        .with_context(|| format!("failed to spawn process {:?}", command))?;
+
+    let output = match stdin_text {
+        None => child.wait_with_output(),
+        // Feeding the command from a second thread, because writing the input and reading the
+        // output have to happen at the same time. Filling up the input pipe while the command
+        // is blocked on an output pipe nobody reads deadlocks both sides.
+        Some(stdin_text) => {
+            let mut stdin = child
+                .stdin
+                .take()
+                .context("failed to open stdin of command")?;
+
+            let (feeding, output) = std::thread::scope(|scope| {
+                // dropping `stdin` when this thread ends closes the pipe, which is how the
+                // command gets to see the end of its input
+                let feeder = scope.spawn(move || stdin.write_all(stdin_text.as_bytes()));
+                let output = child.wait_with_output();
+
+                (feeder.join(), output)
+            });
 
-    let output = command
-        .wait_with_output()
-        .expect("failed to aquire programm output");
+            match feeding {
+                Err(_) => bail!("panic while piping input into command `{}`", command_name),
+                // a command is free to stop reading early, like `head` does. That closes the
+                // pipe under us, which is not an error of its own — the exit code decides.
+                Ok(Err(e)) if e.kind() == std::io::ErrorKind::BrokenPipe => {}
+                Ok(result) => result.context("failed to pipe command into shell")?,
+            }
+
+            output
+        }
+    };
+
+    let output = output.context("failed to aquire programm output")?;
 
     let status: std::process::ExitStatus = output.status;
     if !status.success() {
-        eprintln!("error executing command `{}`.\nProcess terminated with exit code {}.\nProgram output:\n{}",
+        bail!(
+            "error executing command `{}`.\nProcess terminated with exit code {}.\nProgram output:\n{}",
             command_name,
             status,
-            String::from_utf8(output.stderr).unwrap()
+            String::from_utf8_lossy(&output.stderr)
         );
-        std::process::exit(1);
     }
 
-    String::from_utf8(output.stdout).expect("programm output was not valid utf-8")
+    String::from_utf8(output.stdout).context("programm output was not valid utf-8")
 }
 
 #[cfg(test)]
@@ -43,17 +67,73 @@ mod test {
     #[test]
     fn exec1() {
         let input = "printf hello";
-        let result = execute(&[String::from("sh")], Some(input));
+        let result = execute(&[String::from("sh")], Some(input)).unwrap();
         let expected = "hello";
 
         assert_eq!(expected, result);
     }
 
     #[test]
     fn exec2() {
-        let result = execute(&[String::from("printf"), String::from("hello")], None);
+        let result = execute(&[String::from("printf"), String::from("hello")], None).unwrap();
         let expected = "hello";
 
         assert_eq!(expected, result);
     }
+
+    #[test]
+    fn exec_failure_reports_stderr() {
+        let command = [
+            String::from("sh"),
+            String::from("-c"),
+            String::from("echo boom >&2; exit 3"),
+        ];
+
+        let err = execute(&command, None).expect_err("command exits 3, so this must fail");
+        let message = format!("{err:#}");
+
+        assert!(
+            message.contains("boom"),
+            "stderr is missing from: {}",
+            message
+        );
+        assert!(
+            message.contains('3'),
+            "exit code is missing from: {}",
+            message
+        );
+    }
+
+    /// Both pipes hold roughly 64kb, so echoing a megabyte back deadlocks any implementation
+    /// that writes all of stdin before it starts reading stdout.
+    #[test]
+    fn large_input_does_not_deadlock() {
+        let input = "abcdefgh\n".repeat(128 * 1024);
+        let result = execute(&[String::from("cat")], Some(&input)).unwrap();
+
+        assert_eq!(result.len(), input.len());
+        assert_eq!(result, input);
+    }
+
+    /// a command that never reads its input leaves us writing into a closed pipe
+    #[test]
+    fn input_ignored_by_the_command_is_not_an_error() {
+        let input = "abcdefgh\n".repeat(128 * 1024);
+        let command = [
+            String::from("sh"),
+            String::from("-c"),
+            String::from("echo done"),
+        ];
+
+        let result = execute(&command, Some(&input)).unwrap();
+
+        assert_eq!(result, "done\n");
+    }
+
+    #[test]
+    fn exec_failure_of_missing_binary() {
+        let command = [String::from("definitely-not-an-existing-binary-42")];
+
+        assert!(execute(&command, None).is_err());
+    }
 }