Handling a subprocess' output in background threads and accepting user-provided sinks for the output

Question

I have a PR swapping from Command::output() to a custom implementation using threads to handle the subprocess' stdout/stderr while the program is running, instead of waiting for it to complete. The approach was motivated by this answer.

// Executes the given command, capturing its output and exit code in the returned Invocation. // If output_streams is present the output of the command is _also_ written to these streams // concurrently, in order to support displaying a command's output while simultaneously caching // it (instead of waiting for the command to complete before outputting anything). fn execute_subprocess( cmd: impl Into<std::process::Command>, output_streams: Option<(impl Write+Send, impl Write+Send)> ) -> Result<Invocation> { fn maybe_tee(mut source: impl Read, mut sink: Option<impl Write>) -> std::io::Result<Vec<u8>> { let mut ret = Vec::new(); // This initialization can be avoided (safely) once // https://github.com/rust-lang/rust/issues/78485 is stable. let mut buf = [0u8; 1024 * 10]; loop { let num_read = source.read(&mut buf)?; if num_read == 0 { break; } let buf = &buf[..num_read]; if let Some(ref mut sink) = sink { sink.write_all(buf)?; sink.flush()?; } ret.extend(buf); } Ok(ret) } let (out_sink, err_sink) = match output_streams { Some((out, err)) => (Some(out), Some(err)), None => (None, None), }; let mut command: std::process::Command = cmd.into(); use std::process::Stdio; let command = command.stdout(Stdio::piped()).stderr(Stdio::piped()); let start = Instant::now(); let mut child = command.spawn().with_context(|| format!("Failed to run command: {:?}", command))?; let child_out = child.stdout.take().ok_or(anyhow!("cannot attach to child stdout"))?; let child_err = child.stderr.take().ok_or(anyhow!("cannot attach to child stderr"))?; // Using scoped threads means we can take a Write+Send instead of a W+S+'static, allowing // callers to pass mutable references (such as `&mut Vec<u8>`). See also // https://stackoverflow.com/q/32750829/113632 let (stdout, stderr) = std::thread::scope(|s| { let thread_out = s.spawn(|| maybe_tee(child_out, out_sink)); let thread_err = s.spawn(|| maybe_tee(child_err, err_sink)); let stdout = thread_out.join().expect("child stdout thread failed to join").context("stdout pipe failed")?; let stderr = thread_err.join().expect("child stderr thread failed to join").context("stderr pipe failed")?; anyhow::Ok((stdout, stderr)) })?; let status = child.wait()?; let runtime = start.elapsed(); Ok(Invocation { stdout, stderr, // TODO handle signals, see https://stackoverflow.com/q/66272686 exit_code: status.code().unwrap_or(126), runtime, }) } 

I'd appreciate any thoughts on the implementation - or especially the API - of this function¹. In particular:

• Callers are expected to provide a pair of impl Write+Send types, which is compatible with std::io::Stdout/Stderr as well as user-provided types such as &mut Vec<u8> (specifically intended for unit testing).
  • Unfortunately StdoutLock/StderrLock are not Send and therefore cannot be specified here - in principle I believe that would be a better option but I'm not sure how to support that along with other Write impls.
• std::thread::scope is used to simplify the sinks' signature - without scoping I believe it's necessary to use impl Write+Send+'static which prevents using borrowed types like &mut Vec<u8>.
• Trying to avoid .expect() but I'm not sure how to do so with the .join() results, when I swap the expect for ? I hit errors.
• Any general feedback about idiomatic Rust and patterns that could be improved / simplified here.

---

¹ Of course if you're interested in taking a look at the whole PR that'd be greatly appreciated! :)