install: implement copying from streams

src/uu/install/Cargo.toml

@@ -18,16 +18,18 @@ path = "src/install.rs"
 
 [dependencies]
 clap = { workspace = true }
-filetime = { workspace = true }
 file_diff = { workspace = true }
+filetime = { workspace = true }
 libc = { workspace = true }
+nix = { workspace = true }
 uucore = { workspace = true, features = [
   "backup-control",
   "fs",
   "mode",
   "perms",
   "entries",
   "process",
+  "pipes",
 ] }
 
 [[bin]]

src/uu/install/src/install.rs

@@ -10,13 +10,16 @@ mod mode;
 use clap::{crate_version, Arg, ArgAction, ArgMatches, Command};
 use file_diff::diff;
 use filetime::{set_file_times, FileTime};
+
+#[cfg(any(target_os = "linux", target_os = "android"))]
+mod splice;
+
 use std::error::Error;
 use std::fmt::{Debug, Display};
-use std::fs;
 use std::fs::File;
-use std::os::unix::fs::MetadataExt;
-#[cfg(unix)]
-use std::os::unix::prelude::OsStrExt;
+use std::fs::{self, metadata};
+use std::io::{Read, Write};
+use std::os::fd::{AsFd, AsRawFd};
 use std::path::{Path, PathBuf, MAIN_SEPARATOR};
 use std::process;
 use uucore::backup_control::{self, BackupMode};
@@ -29,6 +32,11 @@ use uucore::perms::{wrap_chown, Verbosity, VerbosityLevel};
 use uucore::process::{getegid, geteuid};
 use uucore::{format_usage, help_about, help_usage, show, show_error, show_if_err, uio_error};
 
+#[cfg(unix)]
+use std::os::unix::fs::{FileTypeExt, MetadataExt};
+#[cfg(unix)]
+use std::os::unix::prelude::OsStrExt;
+
 const DEFAULT_MODE: u32 = 0o755;
 const DEFAULT_STRIP_PROGRAM: &str = "strip";
 
@@ -127,6 +135,16 @@ impl Display for InstallError {
     }
 }
 
+#[cfg(unix)]
+pub(crate) trait FdReadable: Read + AsFd + AsRawFd {}
+#[cfg(not(unix))]
+trait FdReadable: Read {}
+
+#[cfg(unix)]
+impl<T> FdReadable for T where T: Read + AsFd + AsRawFd {}
+#[cfg(not(unix))]
+impl<T> FdReadable for T where T: Read {}
+
 #[derive(Clone, Eq, PartialEq)]
 pub enum MainFunction {
     /// Create directories
@@ -736,7 +754,65 @@ fn perform_backup(to: &Path, b: &Behavior) -> UResult<Option<PathBuf>> {
     }
 }
 
-/// Copy a file from one path to another.
+/// Copy a non-special file using std::fs::copy.
+///
+/// # Parameters
+/// * `from` - The source file path.
+/// * `to` - The destination file path.
+///
+/// # Returns
+///
+/// Returns an empty Result or an error in case of failure.
+fn copy_normal_file(from: &Path, to: &Path) -> UResult<()> {
+    if let Err(err) = fs::copy(from, to) {
+        return Err(InstallError::InstallFailed(from.to_path_buf(), to.to_path_buf(), err).into());
+    }
+    Ok(())
+}
+
+/// Read from stream into specified target file.
+///
+/// # Parameters
+/// * `handle` - Open file handle.
+/// * `to` - The destination file path.
+///
+/// # Returns
+///
+/// Returns an empty Result or an error in case of failure.
+fn copy_stream<R: FdReadable>(handle: &mut R, to: &Path) -> UResult<()> {
+    // Overwrite the target file.
+    let mut target_file = File::create(to)?;
+
+    #[cfg(any(target_os = "linux", target_os = "android"))]
+    {
+        // If we're on Linux or Android, try to use the splice() system call
+        // for faster writing. If it works, we're done.
+        if !splice::write_fast_using_splice(handle, &target_file.as_fd())? {
+            return Ok(());
+        }
+    }
+    // If we're not on Linux or Android, or the splice() call failed,
+    // fall back on slower writing.
+    let mut buf = [0; 1024 * 64];
+    while let Ok(n) = handle.read(&mut buf) {
+        if n == 0 {
+            break;
+        }
+        target_file.write_all(&buf[..n])?;
+    }
+
+    // If the splice() call failed and there has been some data written to
+    // stdout via while loop above AND there will be second splice() call
+    // that will succeed, data pushed through splice will be output before
+    // the data buffered in stdout.lock. Therefore additional explicit flush
+    // is required here.
+    target_file.flush()?;
+    Ok(())
+}
+
+/// Copy a
+/// Copy a file from one path to another. Handles the certain cases of special files (e.g character
+/// specials)
 ///
 /// # Parameters
 ///
@@ -760,17 +836,21 @@ fn copy_file(from: &Path, to: &Path) -> UResult<()> {
         }
     }
 
-    if from.as_os_str() == "/dev/null" {
-        /* workaround a limitation of fs::copy
-         * https://github.com/rust-lang/rust/issues/79390
-         */
-        if let Err(err) = File::create(to) {
+    let ft = match metadata(from) {
+        Ok(ft) => ft.file_type(),
+        Err(err) => {
             return Err(
                 InstallError::InstallFailed(from.to_path_buf(), to.to_path_buf(), err).into(),
             );
         }
-    } else if let Err(err) = fs::copy(from, to) {
-        return Err(InstallError::InstallFailed(from.to_path_buf(), to.to_path_buf(), err).into());
+    };
+    match ft {
+        // Stream-based copying to get around the limitations of std::fs::copy
+        ft if ft.is_char_device() || ft.is_block_device() || ft.is_fifo() => {
+            let mut handle = File::open(from)?;
+            copy_stream(&mut handle, to)?;
+        }
+        _ => copy_normal_file(from, to)?,
     }
     Ok(())
 }

src/uu/install/src/splice.rs

@@ -0,0 +1,75 @@
+// This file is part of the uutils coreutils package.
+//
+// For the full copyright and license information, please view the LICENSE
+// file that was distributed with this source code.
+
+use super::{FdReadable, UResult};
+
+use nix::unistd;
+use std::os::{
+    fd::AsFd,
+    unix::io::{AsRawFd, RawFd},
+};
+
+use uucore::pipes::{pipe, splice, splice_exact};
+
+const SPLICE_SIZE: usize = 1024 * 128;
+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 user spaces. 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: FdReadable, S: AsRawFd + AsFd>(
+    handle: &R,
+    write_fd: &S,
+) -> UResult<bool> {
+    let (pipe_rd, pipe_wr) = pipe()?;
+
+    loop {
+        match splice(&handle, &pipe_wr, SPLICE_SIZE) {
+            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.as_raw_fd(), write_fd, n)?;
+                    return Ok(true);
+                }
+            }
+            Err(_) => {
+                return Ok(true);
+            }
+        }
+    }
+}
+
+/// Move exactly `num_bytes` bytes from `read_fd` to `write_fd`.
+///
+/// Panics if not enough bytes can be read.
+fn copy_exact(read_fd: RawFd, write_fd: &impl AsFd, num_bytes: usize) -> nix::Result<()> {
+    let mut left = num_bytes;
+    let mut buf = [0; BUF_SIZE];
+    while left > 0 {
+        let read = unistd::read(read_fd, &mut buf)?;
+        assert_ne!(read, 0, "unexpected end of pipe");
+        let mut written = 0;
+        while written < read {
+            match unistd::write(write_fd, &buf[written..read])? {
+                0 => panic!(),
+                n => written += n,
+            }
+        }
+        left -= read;
+    }
+    Ok(())
+}

tests/by-util/test_install.rs

@@ -1717,3 +1717,53 @@ fn test_install_root_combined() {
     run_and_check(&["-Cv", "c", "d"], "d", 0, 0);
     run_and_check(&["-Cv", "c", "d"], "d", 0, 0);
 }
+
+#[test]
+#[cfg(unix)]
+fn test_install_from_fifo() {
+    use std::fs::OpenOptions;
+    use std::io::Write;
+    use std::thread;
+
+    let pipe_name = "pipe";
+    let target_name = "target";
+    let test_string = "Hello, world!\n";
+
+    let s = TestScenario::new(util_name!());
+    s.fixtures.mkfifo(pipe_name);
+    assert!(s.fixtures.is_fifo(pipe_name));
+
+    let proc = s.ucmd().arg(pipe_name).arg(target_name).run_no_wait();
+
+    let pipe_path = s.fixtures.plus(pipe_name);
+    let thread = thread::spawn(move || {
+        let mut pipe = OpenOptions::new()
+            .write(true)
+            .create(false)
+            .open(pipe_path)
+            .unwrap();
+        pipe.write_all(test_string.as_bytes()).unwrap();
+    });
+
+    proc.wait().unwrap();
+    thread.join().unwrap();
+
+    assert!(s.fixtures.file_exists(target_name));
+    assert_eq!(s.fixtures.read(target_name), test_string);
+}
+
+#[test]
+#[cfg(unix)]
+fn test_install_from_stdin() {
+    let (at, mut ucmd) = at_and_ucmd!();
+    let target = "target";
+    let test_string = "Hello, World!\n";
+
+    ucmd.arg("/dev/fd/0")
+        .arg(target)
+        .pipe_in(test_string)
+        .succeeds();
+
+    assert!(at.file_exists(target));
+    assert_eq!(at.read(target), test_string);
+}