Experiment with fixing wait_with_pipe()

Author: delan

src/child.rs

@@ -115,8 +115,17 @@ impl FunChildren {
         }
     }
 
-    /// Waits for the children processes to exit completely, pipe content will be processed by
-    /// provided function.
+    /// Pipes stdout from the last child in the pipeline to the given function, which runs in
+    /// **the current thread**, then waits for all of the children to exit.
+    ///
+    /// <div class=warning>
+    ///
+    /// # Bugs
+    ///
+    /// The exit status of the last child is **ignored**. If the function returns early, without
+    /// reading from stdout until the last child exits, then the last child may be killed instead
+    /// of being waited for. To avoid these limitations, use [`Self::wait_with_stdout_thread`].
+    /// </div>
     pub fn wait_with_pipe(&mut self, f: &mut dyn FnMut(Box<dyn Read>)) -> CmdResult {
         let child = self.children.pop().unwrap();
         let stderr_thread =
@@ -143,6 +152,22 @@ impl FunChildren {
         CmdChildren::wait_children(&mut self.children)
     }
 
+    /// Pipes stdout from the last child in the pipeline to the given function, which runs in
+    /// **a new thread**, then waits for all of the children to exit.
+    pub fn wait_with_pipe_thread(
+        &mut self,
+        f: impl FnOnce(Box<dyn Read + Send>) + Send + 'static,
+    ) -> CmdResult {
+        if let Some(stdout) = self.children.last_mut().unwrap().stdout.take() {
+            let thread = std::thread::spawn(|| f(Box::new(stdout)));
+            let wait_res = self.wait_with_output().map(|_| ());
+            thread.join().expect("stdout thread panicked");
+            return wait_res;
+        }
+
+        Ok(())
+    }
+
     /// Returns the OS-assigned process identifiers associated with these children processes.
     pub fn pids(&self) -> Vec<u32> {
         self.children.iter().filter_map(|x| x.pid()).collect()

tests/test_macros.rs

@@ -134,7 +134,7 @@ fn test_tls_set() {
 }
 
 #[test]
-fn test_pipe() {
+fn test_pipe() -> CmdResult {
     assert!(run_cmd!(echo "xx").is_ok());
     assert_eq!(run_fun!(echo "xx").unwrap(), "xx");
     assert!(run_cmd!(echo xx | wc).is_ok());
@@ -158,6 +158,64 @@ fn test_pipe() {
     set_pipefail(false);
     assert!(run_fun!(false | true).is_ok());
     set_pipefail(true);
+
+    // test that illustrates the bugs in wait_with_pipe()
+    // FIXME: make set_pipefail() thread safe, then move this to a separate test function
+    assert!(spawn_with_output!(false)?.wait_with_all().0.is_err());
+    assert!(spawn_with_output!(false)?.wait_with_output().is_err());
+    assert!(spawn_with_output!(false)?
+        .wait_with_raw_output(&mut vec![])
+        .is_err());
+
+    // wait_with_pipe() can’t check the exit status of the last child
+    assert!(spawn_with_output!(false)?
+        .wait_with_pipe(&mut |_stdout| {})
+        .is_ok());
+
+    // wait_with_pipe() kills the last child when the provided function returns
+    assert!(spawn_with_output!(sh -c "while :; do :; done")?
+        .wait_with_pipe(&mut |_stdout| {})
+        .is_ok());
+
+    // wait_with_pipe_thread() checks the exit status of the last child, even if pipefail is disabled
+    set_pipefail(false);
+    assert!(spawn_with_output!(true | false)?
+        .wait_with_pipe_thread(|_stdout| {})
+        .is_err());
+    assert!(spawn_with_output!(true | true)?
+        .wait_with_pipe_thread(|_stdout| {})
+        .is_ok());
+    assert!(spawn_with_output!(false)?
+        .wait_with_pipe_thread(|_stdout| {})
+        .is_err());
+    assert!(spawn_with_output!(true)?
+        .wait_with_pipe_thread(|_stdout| {})
+        .is_ok());
+    set_pipefail(true);
+    // wait_with_pipe_thread() checks the exit status of the other children, unless pipefail is disabled
+    set_pipefail(false);
+    assert!(spawn_with_output!(false | true)?
+        .wait_with_pipe_thread(|_stdout| {})
+        .is_ok());
+    set_pipefail(true);
+    assert!(spawn_with_output!(false | true)?
+        .wait_with_pipe_thread(|_stdout| {})
+        .is_err());
+    assert!(spawn_with_output!(true | true)?
+        .wait_with_pipe_thread(|_stdout| {})
+        .is_ok());
+    // wait_with_pipe_thread() handles `ignore`
+    assert!(spawn_with_output!(ignore false | true)?
+        .wait_with_pipe_thread(|_stdout| {})
+        .is_ok());
+    assert!(spawn_with_output!(ignore true | false)?
+        .wait_with_pipe_thread(|_stdout| {})
+        .is_ok());
+    assert!(spawn_with_output!(ignore false)?
+        .wait_with_pipe_thread(|_stdout| {})
+        .is_ok());
+
+    Ok(())
 }
 
 #[test]