Merge pull request #10156 from ChrisDryden/tail-pipe-fix

tail: fix pipe-f test by detecting broken stdout pipe

Author: sylvestre

Cargo.lock

@@ -15,6 +15,8 @@ use std::path::{Path, PathBuf};
 use std::sync::mpsc::{self, Receiver, channel};
 use uucore::display::Quotable;
 use uucore::error::{UResult, USimpleError, set_exit_code};
+#[cfg(target_os = "linux")]
+use uucore::signals::ensure_stdout_not_broken;
 use uucore::translate;
 
 use uucore::show_error;
@@ -160,24 +162,6 @@ impl Observer {
         Ok(())
     }
 
-    pub fn add_stdin(
-        &mut self,
-        display_name: &str,
-        reader: Option<Box<dyn BufRead>>,
-        update_last: bool,
-    ) -> UResult<()> {
-        if self.follow == Some(FollowMode::Descriptor) {
-            return self.add_path(
-                &PathBuf::from(text::DEV_STDIN),
-                display_name,
-                reader,
-                update_last,
-            );
-        }
-
-        Ok(())
-    }
-
     pub fn add_bad_path(
         &mut self,
         path: &Path,
@@ -619,6 +603,11 @@ pub fn follow(mut observer: Observer, settings: &Settings) -> UResult<()> {
             }
             Err(mpsc::RecvTimeoutError::Timeout) => {
                 timeout_counter += 1;
+                // Check if stdout pipe is still open
+                #[cfg(target_os = "linux")]
+                if let Ok(false) = ensure_stdout_not_broken() {
+                    return Ok(());
+                }
             }
             Err(e) => {
                 return Err(USimpleError::new(

src/uu/tail/src/follow/watch.rs

@@ -307,7 +307,6 @@ fn tail_stdin(
             } else {
                 let mut reader = BufReader::new(stdin());
                 unbounded_tail(&mut reader, settings)?;
-                observer.add_stdin(input.display_name.as_str(), Some(Box::new(reader)), true)?;
             }
         }
     }

src/uu/tail/src/tail.rs

@@ -19,7 +19,6 @@ path = "src/tee.rs"
 
 [dependencies]
 clap = { workspace = true }
-nix = { workspace = true, features = ["poll", "fs"] }
 uucore = { workspace = true, features = ["libc", "parser", "signals"] }
 fluent = { workspace = true }
 

src/uu/tee/Cargo.toml

@@ -3,8 +3,6 @@
 // For the full copyright and license information, please view the LICENSE
 // file that was distributed with this source code.
 
-// cSpell:ignore POLLERR POLLRDBAND pfds revents
-
 use clap::{Arg, ArgAction, Command, builder::PossibleValue};
 use std::ffi::OsString;
 use std::fs::OpenOptions;
@@ -18,6 +16,8 @@ use uucore::{format_usage, show_error};
 
 // spell-checker:ignore nopipe
 
+#[cfg(target_os = "linux")]
+use uucore::signals::ensure_stdout_not_broken;
 #[cfg(unix)]
 use uucore::signals::{enable_pipe_errors, ignore_interrupts};
 
@@ -422,45 +422,3 @@ impl Read for NamedReader {
         }
     }
 }
-
-/// Check that if stdout is a pipe, it is not broken.
-#[cfg(target_os = "linux")]
-pub fn ensure_stdout_not_broken() -> Result<bool> {
-    use nix::{
-        poll::{PollFd, PollFlags, PollTimeout},
-        sys::stat::{SFlag, fstat},
-    };
-    use std::os::fd::AsFd;
-
-    let out = stdout();
-
-    // First, check that stdout is a fifo and return true if it's not the case
-    let stat = fstat(out.as_fd())?;
-    if !SFlag::from_bits_truncate(stat.st_mode).contains(SFlag::S_IFIFO) {
-        return Ok(true);
-    }
-
-    // POLLRDBAND is the flag used by GNU tee.
-    let mut pfds = [PollFd::new(out.as_fd(), PollFlags::POLLRDBAND)];
-
-    // Then, ensure that the pipe is not broken.
-    // Use ZERO timeout to return immediately - we just want to check the current state.
-    let res = nix::poll::poll(&mut pfds, PollTimeout::ZERO)?;
-
-    if res > 0 {
-        // poll returned with events ready - check if POLLERR is set (pipe broken)
-        let error = pfds.iter().any(|pfd| {
-            if let Some(revents) = pfd.revents() {
-                revents.contains(PollFlags::POLLERR)
-            } else {
-                true
-            }
-        });
-        return Ok(!error);
-    }
-
-    // res == 0 means no events ready (timeout reached immediately with ZERO timeout).
-    // This means the pipe is healthy (not broken).
-    // res < 0 would be an error, but nix returns Err in that case.
-    Ok(true)
-}

src/uu/tee/src/tee.rs

@@ -93,6 +93,7 @@ nix = { workspace = true, features = [
   "signal",
   "dir",
   "user",
+  "poll",
 ] }
 xattr = { workspace = true, optional = true }
 

src/uucore/Cargo.toml

@@ -3,7 +3,7 @@
 // For the full copyright and license information, please view the LICENSE
 // file that was distributed with this source code.
 
-// spell-checker:ignore (vars/api) fcntl setrlimit setitimer rubout pollable sysconf pgrp
+// spell-checker:ignore (vars/api) fcntl setrlimit setitimer rubout pollable sysconf pgrp pfds revents POLLRDBAND POLLERR
 // spell-checker:ignore (vars/signals) ABRT ALRM CHLD SEGV SIGABRT SIGALRM SIGBUS SIGCHLD SIGCONT SIGDANGER SIGEMT SIGFPE SIGHUP SIGILL SIGINFO SIGINT SIGIO SIGIOT SIGKILL SIGMIGRATE SIGMSG SIGPIPE SIGPRE SIGPROF SIGPWR SIGQUIT SIGSEGV SIGSTOP SIGSYS SIGTALRM SIGTERM SIGTRAP SIGTSTP SIGTHR SIGTTIN SIGTTOU SIGURG SIGUSR SIGVIRT SIGVTALRM SIGWINCH SIGXCPU SIGXFSZ STKFLT PWR THR TSTP TTIN TTOU VIRT VTALRM XCPU XFSZ SIGCLD SIGPOLL SIGWAITING SIGAIOCANCEL SIGLWP SIGFREEZE SIGTHAW SIGCANCEL SIGLOST SIGXRES SIGJVM SIGRTMIN SIGRT SIGRTMAX TALRM AIOCANCEL XRES RTMIN RTMAX LTOSTOP
 
 //! This module provides a way to handle signals in a platform-independent way.
@@ -488,6 +488,48 @@ pub const fn sigpipe_was_ignored() -> bool {
     false
 }
 
+#[cfg(target_os = "linux")]
+pub fn ensure_stdout_not_broken() -> std::io::Result<bool> {
+    use nix::{
+        poll::{PollFd, PollFlags, PollTimeout, poll},
+        sys::stat::{SFlag, fstat},
+    };
+    use std::io::stdout;
+    use std::os::fd::AsFd;
+
+    let out = stdout();
+
+    // First, check that stdout is a fifo and return true if it's not the case
+    let stat = fstat(out.as_fd())?;
+    if !SFlag::from_bits_truncate(stat.st_mode).contains(SFlag::S_IFIFO) {
+        return Ok(true);
+    }
+
+    // POLLRDBAND is the flag used by GNU tee.
+    let mut pfds = [PollFd::new(out.as_fd(), PollFlags::POLLRDBAND)];
+
+    // Then, ensure that the pipe is not broken.
+    // Use ZERO timeout to return immediately - we just want to check the current state.
+    let res = poll(&mut pfds, PollTimeout::ZERO)?;
+
+    if res > 0 {
+        // poll returned with events ready - check if POLLERR is set (pipe broken)
+        let error = pfds.iter().any(|pfd| {
+            if let Some(revents) = pfd.revents() {
+                revents.contains(PollFlags::POLLERR)
+            } else {
+                true
+            }
+        });
+        return Ok(!error);
+    }
+
+    // res == 0 means no events ready (timeout reached immediately with ZERO timeout).
+    // This means the pipe is healthy (not broken).
+    // res < 0 would be an error, but nix returns Err in that case.
+    Ok(true)
+}
+
 #[test]
 fn signal_by_value() {
     assert_eq!(signal_by_name_or_value("0"), Some(0));

src/uucore/src/lib/features/signals.rs

@@ -5039,3 +5039,29 @@ fn tail_n_lines_with_emoji() {
         .succeeds()
         .stdout_only("💐\n");
 }
+
+#[test]
+#[cfg(target_os = "linux")]
+fn test_follow_pipe_f() {
+    new_ucmd!()
+        .args(&["-f", "-c3", "-s.1", "--max-unchanged-stats=1"])
+        .pipe_in("foo\n")
+        .succeeds()
+        .stdout_only("oo\n");
+}
+
+#[test]
+#[cfg(target_os = "linux")]
+fn test_follow_stdout_pipe_close() {
+    let (at, mut ucmd) = at_and_ucmd!();
+    at.write("f", "line1\nline2\n");
+
+    let mut child = ucmd
+        .args(&["-f", "-s.1", "--max-unchanged-stats=1", "f"])
+        .set_stdout(Stdio::piped())
+        .run_no_wait();
+
+    child.stdout_exact_bytes(6); // read "line1\n"
+    child.close_stdout();
+    child.delay(2000).make_assertion().is_not_alive();
+}