cat: Put splice code in separate file, handle more failures (#2067)

* cat: Refactor splice code, handle more failures

* cat: Add tests for stdout redirected to files

src/uu/cat/src/cat.rs

@@ -22,6 +22,12 @@ use std::io::{self, Read, Write};
 use thiserror::Error;
 use uucore::fs::is_stdin_interactive;
 
+/// Linux splice support
+#[cfg(any(target_os = "linux", target_os = "android"))]
+mod splice;
+#[cfg(any(target_os = "linux", target_os = "android"))]
+use std::os::unix::io::{AsRawFd, RawFd};
+
 /// Unix domain socket support
 #[cfg(unix)]
 use std::net::Shutdown;
@@ -30,14 +36,6 @@ use std::os::unix::fs::FileTypeExt;
 #[cfg(unix)]
 use unix_socket::UnixStream;
 
-/// Linux splice support
-#[cfg(any(target_os = "linux", target_os = "android"))]
-use nix::fcntl::{splice, SpliceFFlags};
-#[cfg(any(target_os = "linux", target_os = "android"))]
-use nix::unistd::pipe;
-#[cfg(any(target_os = "linux", target_os = "android"))]
-use std::os::unix::io::{AsRawFd, RawFd};
-
 static NAME: &str = "cat";
 static VERSION: &str = env!("CARGO_PKG_VERSION");
 static SYNTAX: &str = "[OPTION]... [FILE]...";
@@ -395,7 +393,7 @@ fn write_fast<R: Read>(handle: &mut InputHandle<R>) -> CatResult<()> {
     {
         // If we're on Linux or Android, try to use the splice() system call
         // for faster writing. If it works, we're done.
-        if !write_fast_using_splice(handle, stdout_lock.as_raw_fd())? {
+        if !splice::write_fast_using_splice(handle, stdout_lock.as_raw_fd())? {
             return Ok(());
         }
     }
@@ -411,75 +409,6 @@ fn write_fast<R: Read>(handle: &mut InputHandle<R>) -> CatResult<()> {
     Ok(())
 }
 
-/// This function is called from `write_fast()` on Linux and Android. The
-/// function `splice()` is used to move data between two file descriptors
-/// without copying between kernel- and userspace. This results in a large
-/// speedup.
-///
-/// The `bool` in the result value indicates if we need to fall back to normal
-/// copying or not. False means we don't have to.
-#[cfg(any(target_os = "linux", target_os = "android"))]
-#[inline]
-fn write_fast_using_splice<R: Read>(handle: &mut InputHandle<R>, writer: RawFd) -> CatResult<bool> {
-    const BUF_SIZE: usize = 1024 * 16;
-
-    let (pipe_rd, pipe_wr) = pipe()?;
-
-    // We only fall back if splice fails on the first call.
-    match splice(
-        handle.file_descriptor,
-        None,
-        pipe_wr,
-        None,
-        BUF_SIZE,
-        SpliceFFlags::empty(),
-    ) {
-        Ok(n) => {
-            if n == 0 {
-                return Ok(false);
-            }
-            splice_exact(pipe_rd, writer, n)?;
-        }
-        Err(_) => {
-            return Ok(true);
-        }
-    }
-
-    loop {
-        let n = splice(
-            handle.file_descriptor,
-            None,
-            pipe_wr,
-            None,
-            BUF_SIZE,
-            SpliceFFlags::empty(),
-        )?;
-        if n == 0 {
-            // We read 0 bytes from the input,
-            // which means we're done copying.
-            break;
-        }
-        splice_exact(pipe_rd, writer, n)?;
-    }
-
-    Ok(false)
-}
-
-/// Splice wrapper which handles short writes
-#[cfg(any(target_os = "linux", target_os = "android"))]
-#[inline]
-fn splice_exact(read_fd: RawFd, write_fd: RawFd, num_bytes: usize) -> nix::Result<()> {
-    let mut left = num_bytes;
-    loop {
-        let written = splice(read_fd, None, write_fd, None, left, SpliceFFlags::empty())?;
-        left -= written;
-        if left == 0 {
-            break;
-        }
-    }
-    Ok(())
-}
-
 /// Outputs file contents to stdout in a line-by-line fashion,
 /// propagating any errors that might occur.
 fn write_lines<R: Read>(

src/uu/cat/src/splice.rs

@@ -0,0 +1,91 @@
+use super::{CatResult, InputHandle};
+
+use nix::fcntl::{splice, SpliceFFlags};
+use nix::unistd::{self, pipe};
+use std::io::Read;
+use std::os::unix::io::RawFd;
+
+const BUF_SIZE: usize = 1024 * 16;
+
+/// This function is called from `write_fast()` on Linux and Android. The
+/// function `splice()` is used to move data between two file descriptors
+/// without copying between kernel- and userspace. This results in a large
+/// speedup.
+///
+/// The `bool` in the result value indicates if we need to fall back to normal
+/// copying or not. False means we don't have to.
+#[inline]
+pub(super) fn write_fast_using_splice<R: Read>(
+    handle: &mut InputHandle<R>,
+    write_fd: RawFd,
+) -> CatResult<bool> {
+    let (pipe_rd, pipe_wr) = match pipe() {
+        Ok(r) => r,
+        Err(_) => {
+            // It is very rare that creating a pipe fails, but it can happen.
+            return Ok(true);
+        }
+    };
+
+    loop {
+        match splice(
+            handle.file_descriptor,
+            None,
+            pipe_wr,
+            None,
+            BUF_SIZE,
+            SpliceFFlags::empty(),
+        ) {
+            Ok(n) => {
+                if n == 0 {
+                    return Ok(false);
+                }
+                if splice_exact(pipe_rd, write_fd, n).is_err() {
+                    // If the first splice manages to copy to the intermediate
+                    // pipe, but the second splice to stdout fails for some reason
+                    // we can recover by copying the data that we have from the
+                    // intermediate pipe to stdout using normal read/write. Then
+                    // we tell the caller to fall back.
+                    copy_exact(pipe_rd, write_fd, n)?;
+                    return Ok(true);
+                }
+            }
+            Err(_) => {
+                return Ok(true);
+            }
+        }
+    }
+}
+
+/// Splice wrapper which handles short writes.
+#[inline]
+fn splice_exact(read_fd: RawFd, write_fd: RawFd, num_bytes: usize) -> nix::Result<()> {
+    let mut left = num_bytes;
+    loop {
+        let written = splice(read_fd, None, write_fd, None, left, SpliceFFlags::empty())?;
+        left -= written;
+        if left == 0 {
+            break;
+        }
+    }
+    Ok(())
+}
+
+/// Caller must ensure that `num_bytes <= BUF_SIZE`, otherwise this function
+/// will panic. The way we use this function in `write_fast_using_splice`
+/// above is safe because `splice` is set to write at most `BUF_SIZE` to the
+/// pipe.
+#[inline]
+fn copy_exact(read_fd: RawFd, write_fd: RawFd, num_bytes: usize) -> nix::Result<()> {
+    let mut left = num_bytes;
+    let mut buf = [0; BUF_SIZE];
+    loop {
+        let read = unistd::read(read_fd, &mut buf[..left])?;
+        let written = unistd::write(write_fd, &mut buf[..read])?;
+        left -= written;
+        if left == 0 {
+            break;
+        }
+    }
+    Ok(())
+}

tests/by-util/test_cat.rs

@@ -1,4 +1,7 @@
 use crate::common::util::*;
+#[cfg(unix)]
+use std::fs::OpenOptions;
+#[cfg(unix)]
 use std::io::Read;
 
 #[test]
@@ -54,7 +57,6 @@ fn test_no_options_big_input() {
 #[test]
 #[cfg(unix)]
 fn test_fifo_symlink() {
-    use std::fs::OpenOptions;
     use std::io::Write;
     use std::thread;
 
@@ -85,6 +87,74 @@ fn test_fifo_symlink() {
     thread.join().unwrap();
 }
 
+#[test]
+#[cfg(unix)]
+fn test_piped_to_regular_file() {
+    use std::fs::read_to_string;
+
+    for &append in &[true, false] {
+        let s = TestScenario::new(util_name!());
+        let file_path = s.fixtures.plus("file.txt");
+
+        {
+            let file = OpenOptions::new()
+                .create_new(true)
+                .write(true)
+                .append(append)
+                .open(&file_path)
+                .unwrap();
+
+            s.ucmd()
+                .set_stdout(file)
+                .pipe_in_fixture("alpha.txt")
+                .succeeds();
+        }
+        let contents = read_to_string(&file_path).unwrap();
+        assert_eq!(contents, "abcde\nfghij\nklmno\npqrst\nuvwxyz\n");
+    }
+}
+
+#[test]
+#[cfg(unix)]
+fn test_piped_to_dev_null() {
+    for &append in &[true, false] {
+        let s = TestScenario::new(util_name!());
+        {
+            let dev_null = OpenOptions::new()
+                .write(true)
+                .append(append)
+                .open("/dev/null")
+                .unwrap();
+
+            s.ucmd()
+                .set_stdout(dev_null)
+                .pipe_in_fixture("alpha.txt")
+                .succeeds();
+        }
+    }
+}
+
+#[test]
+#[cfg(any(target_os = "linux", target_os = "freebsd", target_os = "netbsd"))]
+fn test_piped_to_dev_full() {
+    for &append in &[true, false] {
+        let s = TestScenario::new(util_name!());
+        {
+            let dev_full = OpenOptions::new()
+                .write(true)
+                .append(append)
+                .open("/dev/full")
+                .unwrap();
+
+            s.ucmd()
+                .set_stdout(dev_full)
+                .pipe_in_fixture("alpha.txt")
+                .fails()
+                .stderr_contains(&"No space left on device".to_owned());
+        }
+    }
+}
+
 #[test]
 fn test_directory() {
     let s = TestScenario::new(util_name!());

tests/common/util.rs

@@ -696,8 +696,11 @@ pub struct UCommand {
     comm_string: String,
     tmpd: Option<Rc<TempDir>>,
     has_run: bool,
-    stdin: Option<Vec<u8>>,
     ignore_stdin_write_error: bool,
+    stdin: Option<Stdio>,
+    stdout: Option<Stdio>,
+    stderr: Option<Stdio>,
+    bytes_into_stdin: Option<Vec<u8>>,
 }
 
 impl UCommand {
@@ -726,8 +729,11 @@ impl UCommand {
                 cmd
             },
             comm_string: String::from(arg.as_ref().to_str().unwrap()),
-            stdin: None,
             ignore_stdin_write_error: false,
+            bytes_into_stdin: None,
+            stdin: None,
+            stdout: None,
+            stderr: None,
         }
     }
 
@@ -738,6 +744,21 @@ impl UCommand {
         ucmd
     }
 
+    pub fn set_stdin<T: Into<Stdio>>(&mut self, stdin: T) -> &mut UCommand {
+        self.stdin = Some(stdin.into());
+        self
+    }
+
+    pub fn set_stdout<T: Into<Stdio>>(&mut self, stdout: T) -> &mut UCommand {
+        self.stdout = Some(stdout.into());
+        self
+    }
+
+    pub fn set_stderr<T: Into<Stdio>>(&mut self, stderr: T) -> &mut UCommand {
+        self.stderr = Some(stderr.into());
+        self
+    }
+
     /// Add a parameter to the invocation. Path arguments are treated relative
     /// to the test environment directory.
     pub fn arg<S: AsRef<OsStr>>(&mut self, arg: S) -> &mut UCommand {
@@ -767,10 +788,10 @@ impl UCommand {
 
     /// provides stdinput to feed in to the command when spawned
     pub fn pipe_in<T: Into<Vec<u8>>>(&mut self, input: T) -> &mut UCommand {
-        if self.stdin.is_some() {
+        if self.bytes_into_stdin.is_some() {
             panic!("{}", MULTIPLE_STDIN_MEANINGLESS);
         }
-        self.stdin = Some(input.into());
+        self.bytes_into_stdin = Some(input.into());
         self
     }
 
@@ -784,7 +805,7 @@ impl UCommand {
     /// This is typically useful to test non-standard workflows
     /// like feeding something to a command that does not read it
     pub fn ignore_stdin_write_error(&mut self) -> &mut UCommand {
-        if self.stdin.is_none() {
+        if self.bytes_into_stdin.is_none() {
             panic!("{}", NO_STDIN_MEANINGLESS);
         }
         self.ignore_stdin_write_error = true;
@@ -813,13 +834,13 @@ impl UCommand {
         log_info("run", &self.comm_string);
         let mut child = self
             .raw
-            .stdin(Stdio::piped())
-            .stdout(Stdio::piped())
-            .stderr(Stdio::piped())
+            .stdin(self.stdin.take().unwrap_or_else(|| Stdio::piped()))
+            .stdout(self.stdout.take().unwrap_or_else(|| Stdio::piped()))
+            .stderr(self.stderr.take().unwrap_or_else(|| Stdio::piped()))
             .spawn()
             .unwrap();
 
-        if let Some(ref input) = self.stdin {
+        if let Some(ref input) = self.bytes_into_stdin {
             let write_result = child
                 .stdin
                 .take()