Don't read stderr into memory when not captured

Cargo.toml

@@ -11,7 +11,7 @@ keywords = ["child", "child-process", "command", "process", "shell"]
 categories = ["filesystem", "os"]
 
 [workspace]
-members = [".", "memory-test"]
+members = [".", "memory-tests"]
 
 [dependencies]
 rustversion = "1.0.4"

memory-test/Cargo.toml → memory-tests/Cargo.toml

@@ -1,5 +1,5 @@
 [package]
-name = "memory-test"
+name = "memory-tests"
 version = "0.0.0"
 edition = "2018"
 

memory-tests/src/bin/cradle_user.rs

@@ -0,0 +1,13 @@
+use cradle::prelude::*;
+
+fn main() {
+    let mut args = std::env::args();
+    let stream_type: String = args.nth(1).unwrap();
+    let bytes: usize = args.next().unwrap().parse().unwrap();
+    eprintln!("consuming {} KiB", bytes / 2_usize.pow(10));
+    cmd_unit!(
+        "./target/release/produce_bytes",
+        stream_type,
+        bytes.to_string()
+    );
+}

memory-tests/src/bin/produce_bytes.rs

@@ -0,0 +1,22 @@
+use anyhow::Result;
+use std::io::{self, Write};
+
+fn main() -> Result<()> {
+    let mut args = std::env::args();
+    let stream_type: String = args.nth(1).unwrap();
+    let mut bytes: usize = args.next().unwrap().parse()?;
+    eprintln!("writing {} KiB to {}", bytes / 2_usize.pow(10), stream_type);
+    let buffer = &[b'x'; 1024];
+    let mut stream: Box<dyn Write> = match stream_type.as_str() {
+        "stdout" => Box::new(io::stdout()),
+        "stderr" => Box::new(io::stderr()),
+        _ => panic!("unknown stream type: {}", stream_type),
+    };
+    while bytes > 0 {
+        let chunk_size = bytes.min(1024);
+        stream.write_all(&buffer[..chunk_size])?;
+        bytes -= chunk_size;
+    }
+    stream.flush()?;
+    Ok(())
+}

memory-test/src/bin/run_test.rs → memory-tests/src/bin/run.rs

@@ -1,35 +1,41 @@
 use anyhow::Result;
-use cradle::prelude::*;
 use std::process::{Command, Stdio};
 
 fn from_mib(mebibytes: usize) -> usize {
     mebibytes * 2_usize.pow(20)
 }
 
 fn main() -> Result<()> {
-    Split("cargo build --release").run_unit();
+    test("stdout")?;
+    test("stderr")?;
+    Ok(())
+}
+
+fn test(stream_type: &str) -> Result<()> {
     let bytes = from_mib(64);
-    let memory_consumption = measure_memory_consumption(bytes)?;
+    let memory_consumption = measure_memory_consumption(stream_type, bytes)?;
     let allowed_memory_consumption = from_mib(16);
     assert!(
         memory_consumption < allowed_memory_consumption,
-        "Maximum resident set size: {}, allowed upper limit: {}",
+        "stream type: {}, maximum resident set size: {}, allowed upper limit: {}",
+        stream_type,
         memory_consumption,
         allowed_memory_consumption
     );
     Ok(())
 }
 
-fn measure_memory_consumption(bytes: usize) -> Result<usize> {
+fn measure_memory_consumption(stream_type: &str, bytes: usize) -> Result<usize> {
     let output = Command::new("/usr/bin/time")
         .arg("-v")
         .arg("./target/release/cradle_user")
+        .arg(stream_type)
         .arg(bytes.to_string())
         .stdout(Stdio::null())
         .output()?;
     let stderr = String::from_utf8(output.stderr)?;
-    eprintln!("{}", stderr);
     if !output.status.success() {
+        eprintln!("{}", stderr);
         panic!("running 'cradle_user' failed");
     }
     let memory_size_prefix = "Maximum resident set size (kbytes): ";

src/collected_output.rs

@@ -9,16 +9,44 @@ use std::{
 pub(crate) struct Waiter {
     stdin: JoinHandle<io::Result<()>>,
     stdout: JoinHandle<io::Result<Option<Vec<u8>>>>,
-    stderr: JoinHandle<io::Result<Vec<u8>>>,
+    stderr: JoinHandle<io::Result<Option<Vec<u8>>>>,
 }
 
 impl Waiter {
+    fn spawn_standard_stream_handler(
+        capture_stream: bool,
+        mut source: impl Read + Send + 'static,
+        mut relay_sink: impl Write + Send + 'static,
+    ) -> JoinHandle<io::Result<Option<Vec<u8>>>> {
+        thread::spawn(move || -> io::Result<Option<Vec<u8>>> {
+            let mut collected = if capture_stream {
+                Some(Vec::new())
+            } else {
+                None
+            };
+            let buffer = &mut [0; 256];
+            loop {
+                let length = source.read(buffer)?;
+                if (length) == 0 {
+                    break;
+                }
+                if let Some(collected) = &mut collected {
+                    collected.extend(&buffer[..length]);
+                }
+                if !capture_stream {
+                    relay_sink.write_all(&buffer[..length])?;
+                }
+            }
+            Ok(collected)
+        })
+    }
+
     pub(crate) fn spawn_standard_stream_relaying<Stdout, Stderr>(
         context: &Context<Stdout, Stderr>,
         config: &Config,
         mut child_stdin: ChildStdin,
-        mut child_stdout: ChildStdout,
-        mut child_stderr: ChildStderr,
+        child_stdout: ChildStdout,
+        child_stderr: ChildStderr,
     ) -> Self
     where
         Stdout: Write + Send + Clone + 'static,
@@ -29,46 +57,16 @@ impl Waiter {
             child_stdin.write_all(&config_stdin)?;
             Ok(())
         });
-        let mut context_clone = context.clone();
-        let capture_stdout = config.capture_stdout;
-        let stdout_join_handle = thread::spawn(move || -> io::Result<Option<Vec<u8>>> {
-            let mut collected_stdout = if capture_stdout {
-                Some(Vec::new())
-            } else {
-                None
-            };
-            let buffer = &mut [0; 256];
-            loop {
-                let length = child_stdout.read(buffer)?;
-                if (length) == 0 {
-                    break;
-                }
-                if let Some(collected_stdout) = &mut collected_stdout {
-                    collected_stdout.extend(&buffer[..length]);
-                }
-                if !capture_stdout {
-                    context_clone.stdout.write_all(&buffer[..length])?;
-                }
-            }
-            Ok(collected_stdout)
-        });
-        let mut context_clone = context.clone();
-        let capture_stderr = config.capture_stderr;
-        let stderr_join_handle = thread::spawn(move || -> io::Result<Vec<u8>> {
-            let mut collected_stderr = Vec::new();
-            let buffer = &mut [0; 256];
-            loop {
-                let length = child_stderr.read(buffer)?;
-                if (length) == 0 {
-                    break;
-                }
-                collected_stderr.extend(&buffer[..length]);
-                if !capture_stderr {
-                    context_clone.stderr.write_all(&buffer[..length])?;
-                }
-            }
-            Ok(collected_stderr)
-        });
+        let stdout_join_handle = Self::spawn_standard_stream_handler(
+            config.capture_stdout,
+            child_stdout,
+            context.stdout.clone(),
+        );
+        let stderr_join_handle = Self::spawn_standard_stream_handler(
+            config.capture_stderr,
+            child_stderr,
+            context.stderr.clone(),
+        );
         Waiter {
             stdin: stdin_join_handle,
             stdout: stdout_join_handle,
@@ -96,5 +94,5 @@ impl Waiter {
 #[derive(Debug)]
 pub(crate) struct CollectedOutput {
     pub(crate) stdout: Option<Vec<u8>>,
-    pub(crate) stderr: Vec<u8>,
+    pub(crate) stderr: Option<Vec<u8>>,
 }

src/error.rs

@@ -27,6 +27,7 @@ pub enum Error {
         source: Arc<FromUtf8Error>,
     },
     Internal {
+        message: String,
         full_command: String,
         config: Config,
     },
@@ -40,8 +41,9 @@ impl Error {
         }
     }
 
-    pub(crate) fn internal(config: &Config) -> Error {
+    pub(crate) fn internal(message: &str, config: &Config) -> Error {
         Error::Internal {
+            message: message.to_string(),
             full_command: config.full_command(),
             config: config.clone(),
         }

src/lib.rs

@@ -323,7 +323,7 @@ where
 #[derive(Clone, Debug)]
 pub struct RunResult {
     stdout: Option<Vec<u8>>,
-    stderr: Vec<u8>,
+    stderr: Option<Vec<u8>>,
     exit_status: ExitStatus,
 }
 

src/output.rs

@@ -170,8 +170,9 @@ impl Output for StdoutUntrimmed {
 
     #[doc(hidden)]
     fn from_run_result(config: &Config, result: Result<RunResult, Error>) -> Result<Self, Error> {
-        let result = result?;
-        let stdout = result.stdout.ok_or_else(|| Error::internal(config))?;
+        let stdout = result?
+            .stdout
+            .ok_or_else(|| Error::internal("stdout not captured", config))?;
         Ok(StdoutUntrimmed(String::from_utf8(stdout).map_err(
             |source| Error::InvalidUtf8ToStdout {
                 full_command: config.full_command(),
@@ -209,12 +210,15 @@ impl Output for Stderr {
 
     #[doc(hidden)]
     fn from_run_result(config: &Config, result: Result<RunResult, Error>) -> Result<Self, Error> {
-        Ok(Stderr(String::from_utf8(result?.stderr).map_err(
-            |source| Error::InvalidUtf8ToStderr {
+        let stderr = result?
+            .stderr
+            .ok_or_else(|| Error::internal("stderr not captured", config))?;
+        Ok(Stderr(String::from_utf8(stderr).map_err(|source| {
+            Error::InvalidUtf8ToStderr {
                 full_command: config.full_command(),
                 source: Arc::new(source),
-            },
-        )?))
+            }
+        })?))
     }
 }
 

tests/integration.rs

@@ -318,6 +318,6 @@ mod run_interface {
 #[test]
 fn memory_test() {
     use cradle::prelude::*;
-    cmd_unit!(%"cargo build -p memory-test --release");
-    cmd_unit!(%"cargo run -p memory-test --bin run_test");
+    cmd_unit!(%"cargo build -p memory-tests --release");
+    cmd_unit!(%"cargo run -p memory-tests --bin run");
 }