Merge pull request #4164 from jfinkels/dd-fifo-seek

dd: allow skipping and seeking in FIFOs

src/uu/dd/src/dd.rs

@@ -27,7 +27,9 @@ use std::cmp;
 use std::env;
 use std::ffi::OsString;
 use std::fs::{File, OpenOptions};
-use std::io::{self, Read, Seek, SeekFrom, Stdout, Write};
+use std::io::{self, Read, Seek, SeekFrom, Stdin, Stdout, Write};
+#[cfg(unix)]
+use std::os::unix::fs::FileTypeExt;
 #[cfg(any(target_os = "linux", target_os = "android"))]
 use std::os::unix::fs::OpenOptionsExt;
 use std::path::Path;
@@ -90,30 +92,106 @@ impl Num {
     }
 }
 
-struct Input<'a, R: Read> {
+/// Data sources.
-    src: R,
+enum Source {
+    /// Input from stdin.
+    Stdin(Stdin),
+
+    /// Input from a file.
+    File(File),
+
+    /// Input from a named pipe, also known as a FIFO.
+    #[cfg(unix)]
+    Fifo(File),
+}
+
+impl Source {
+    fn skip(&mut self, n: u64) -> io::Result<u64> {
+        match self {
+            Self::Stdin(stdin) => match io::copy(&mut stdin.take(n), &mut io::sink()) {
+                Ok(m) if m < n => {
+                    show_error!("'standard input': cannot skip to specified offset");
+                    Ok(m)
+                }
+                Ok(m) => Ok(m),
+                Err(e) => Err(e),
+            },
+            Self::File(f) => f.seek(io::SeekFrom::Start(n)),
+            #[cfg(unix)]
+            Self::Fifo(f) => io::copy(&mut f.take(n), &mut io::sink()),
+        }
+    }
+}
+
+impl Read for Source {
+    fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
+        match self {
+            Self::Stdin(stdin) => stdin.read(buf),
+            Self::File(f) => f.read(buf),
+            #[cfg(unix)]
+            Self::Fifo(f) => f.read(buf),
+        }
+    }
+}
+
+/// The source of the data, configured with the given settings.
+///
+/// Use the [`Input::new_stdin`] or [`Input::new_file`] functions to
+/// construct a new instance of this struct. Then pass the instance to
+/// the [`Output::dd_out`] function to execute the main copy operation
+/// for `dd`.
+struct Input<'a> {
+    /// The source from which bytes will be read.
+    src: Source,
+
+    /// Configuration settings for how to read the data.
     settings: &'a Settings,
 }
 
-impl<'a> Input<'a, io::Stdin> {
+impl<'a> Input<'a> {
-    fn new(settings: &'a Settings) -> UResult<Self> {
+    /// Instantiate this struct with stdin as a source.
-        let mut input = Self {
+    fn new_stdin(settings: &'a Settings) -> UResult<Self> {
-            src: io::stdin(),
+        let mut src = Source::Stdin(io::stdin());
-            settings,
+        if settings.skip > 0 {
+            src.skip(settings.skip)?;
+        }
+        Ok(Self { src, settings })
+    }
+
+    /// Instantiate this struct with the named file as a source.
+    fn new_file(filename: &Path, settings: &'a Settings) -> UResult<Self> {
+        let src = {
+            let mut opts = OpenOptions::new();
+            opts.read(true);
+
+            #[cfg(any(target_os = "linux", target_os = "android"))]
+            if let Some(libc_flags) = make_linux_iflags(&settings.iflags) {
+                opts.custom_flags(libc_flags);
+            }
+
+            opts.open(filename)
+                .map_err_context(|| format!("failed to open {}", filename.quote()))?
         };
 
+        let mut src = Source::File(src);
         if settings.skip > 0 {
-            if let Err(e) = input.read_skip(settings.skip) {
+            src.skip(settings.skip)?;
-                if let io::ErrorKind::UnexpectedEof = e.kind() {
-                    show_error!("'standard input': cannot skip to specified offset");
-                } else {
-                    return io::Result::Err(e)
-                        .map_err_context(|| "I/O error while skipping".to_string());
-                }
-            }
         }
+        Ok(Self { src, settings })
+    }
 
-        Ok(input)
+    /// Instantiate this struct with the named pipe as a source.
+    #[cfg(unix)]
+    fn new_fifo(filename: &Path, settings: &'a Settings) -> UResult<Self> {
+        let mut opts = OpenOptions::new();
+        opts.read(true);
+        #[cfg(any(target_os = "linux", target_os = "android"))]
+        opts.custom_flags(make_linux_iflags(&settings.iflags).unwrap_or(0));
+        let mut src = Source::Fifo(opts.open(filename)?);
+        if settings.skip > 0 {
+            src.skip(settings.skip)?;
+        }
+        Ok(Self { src, settings })
     }
 }
 
@@ -153,31 +231,7 @@ fn make_linux_iflags(iflags: &IFlags) -> Option<libc::c_int> {
     }
 }
 
-impl<'a> Input<'a, File> {
+impl<'a> Read for Input<'a> {
-    fn new(filename: &Path, settings: &'a Settings) -> UResult<Self> {
-        let mut src = {
-            let mut opts = OpenOptions::new();
-            opts.read(true);
-
-            #[cfg(any(target_os = "linux", target_os = "android"))]
-            if let Some(libc_flags) = make_linux_iflags(&settings.iflags) {
-                opts.custom_flags(libc_flags);
-            }
-
-            opts.open(filename)
-                .map_err_context(|| format!("failed to open {}", filename.quote()))?
-        };
-
-        if settings.skip > 0 {
-            src.seek(io::SeekFrom::Start(settings.skip))
-                .map_err_context(|| "failed to seek in input file".to_string())?;
-        }
-
-        Ok(Self { src, settings })
-    }
-}
-
-impl<'a, R: Read> Read for Input<'a, R> {
     fn read(&mut self, buf: &mut [u8]) -> io::Result<usize> {
         let mut base_idx = 0;
         let target_len = buf.len();
@@ -200,7 +254,7 @@ impl<'a, R: Read> Read for Input<'a, R> {
     }
 }
 
-impl<'a, R: Read> Input<'a, R> {
+impl<'a> Input<'a> {
     /// Fills a given buffer.
     /// Reads in increments of 'self.ibs'.
     /// The start of each ibs-sized read follows the previous one.
@@ -266,20 +320,6 @@ impl<'a, R: Read> Input<'a, R> {
             records_truncated: 0,
         })
     }
-
-    /// Skips amount_to_read bytes from the Input by copying into a sink
-    fn read_skip(&mut self, amount_to_read: u64) -> std::io::Result<()> {
-        let copy_result = io::copy(&mut self.src.by_ref().take(amount_to_read), &mut io::sink());
-        if let Ok(n) = copy_result {
-            if n != amount_to_read {
-                io::Result::Err(io::Error::new(io::ErrorKind::UnexpectedEof, ""))
-            } else {
-                Ok(())
-            }
-        } else {
-            io::Result::Err(copy_result.unwrap_err())
-        }
-    }
 }
 
 enum Density {
@@ -297,6 +337,14 @@ enum Dest {
     /// The [`Density`] component indicates whether to attempt to
     /// write a sparse file when all-zero blocks are encountered.
     File(File, Density),
+
+    /// Output to a named pipe, also known as a FIFO.
+    #[cfg(unix)]
+    Fifo(File),
+
+    /// Output to nothing, dropping each byte written to the output.
+    #[cfg(unix)]
+    Sink,
 }
 
 impl Dest {
@@ -307,6 +355,13 @@ impl Dest {
                 f.flush()?;
                 f.sync_all()
             }
+            #[cfg(unix)]
+            Self::Fifo(f) => {
+                f.flush()?;
+                f.sync_all()
+            }
+            #[cfg(unix)]
+            Self::Sink => Ok(()),
         }
     }
 
@@ -317,6 +372,13 @@ impl Dest {
                 f.flush()?;
                 f.sync_data()
             }
+            #[cfg(unix)]
+            Self::Fifo(f) => {
+                f.flush()?;
+                f.sync_data()
+            }
+            #[cfg(unix)]
+            Self::Sink => Ok(()),
         }
     }
 
@@ -324,17 +386,24 @@ impl Dest {
         match self {
             Self::Stdout(stdout) => io::copy(&mut io::repeat(0).take(n), stdout),
             Self::File(f, _) => f.seek(io::SeekFrom::Start(n)),
+            #[cfg(unix)]
+            Self::Fifo(f) => {
+                // Seeking in a named pipe means *reading* from the pipe.
+                io::copy(&mut f.take(n), &mut io::sink())
+            }
+            #[cfg(unix)]
+            Self::Sink => Ok(0),
         }
     }
 
     /// Truncate the underlying file to the current stream position, if possible.
     fn truncate(&mut self) -> io::Result<()> {
         match self {
-            Self::Stdout(_) => Ok(()),
             Self::File(f, _) => {
                 let pos = f.stream_position()?;
                 f.set_len(pos)
             }
+            _ => Ok(()),
         }
     }
 }
@@ -357,6 +426,10 @@ impl Write for Dest {
             }
             Self::File(f, _) => f.write(buf),
             Self::Stdout(stdout) => stdout.write(buf),
+            #[cfg(unix)]
+            Self::Fifo(f) => f.write(buf),
+            #[cfg(unix)]
+            Self::Sink => Ok(buf.len()),
         }
     }
 
@@ -364,6 +437,10 @@ impl Write for Dest {
         match self {
             Self::Stdout(stdout) => stdout.flush(),
             Self::File(f, _) => f.flush(),
+            #[cfg(unix)]
+            Self::Fifo(f) => f.flush(),
+            #[cfg(unix)]
+            Self::Sink => Ok(()),
         }
     }
 }
@@ -433,6 +510,35 @@ impl<'a> Output<'a> {
         Ok(Self { dst, settings })
     }
 
+    /// Instantiate this struct with the given named pipe as a destination.
+    #[cfg(unix)]
+    fn new_fifo(filename: &Path, settings: &'a Settings) -> UResult<Self> {
+        // We simulate seeking in a FIFO by *reading*, so we open the
+        // file for reading. But then we need to close the file and
+        // re-open it for writing.
+        if settings.seek > 0 {
+            Dest::Fifo(File::open(filename)?).seek(settings.seek)?;
+        }
+        // If `count=0`, then we don't bother opening the file for
+        // writing because that would cause this process to block
+        // indefinitely.
+        if let Some(Num::Blocks(0) | Num::Bytes(0)) = settings.count {
+            let dst = Dest::Sink;
+            return Ok(Self { dst, settings });
+        }
+        // At this point, we know there is at least one block to write
+        // to the output, so we open the file for writing.
+        let mut opts = OpenOptions::new();
+        opts.write(true)
+            .create(!settings.oconv.nocreat)
+            .create_new(settings.oconv.excl)
+            .append(settings.oflags.append);
+        #[cfg(any(target_os = "linux", target_os = "android"))]
+        opts.custom_flags(make_linux_oflags(&settings.oflags).unwrap_or(0));
+        let dst = Dest::Fifo(opts.open(filename)?);
+        Ok(Self { dst, settings })
+    }
+
     /// Write the given bytes one block at a time.
     ///
     /// This may write partial blocks (for example, if the underlying
@@ -485,7 +591,7 @@ impl<'a> Output<'a> {
     ///
     /// If there is a problem reading from the input or writing to
     /// this output.
-    fn dd_out<R: Read>(mut self, mut i: Input<R>) -> std::io::Result<()> {
+    fn dd_out(mut self, mut i: Input) -> std::io::Result<()> {
         // The read and write statistics.
         //
         // These objects are counters, initialized to zero. After each
@@ -645,12 +751,13 @@ fn make_linux_oflags(oflags: &OFlags) -> Option<libc::c_int> {
     }
 }
 
-/// Read helper performs read operations common to all dd reads, and dispatches the buffer to relevant helper functions as dictated by the operations requested by the user.
+/// Read from an input (that is, a source of bytes) into the given buffer.
-fn read_helper<R: Read>(
+///
-    i: &mut Input<R>,
+/// This function also performs any conversions as specified by
-    buf: &mut Vec<u8>,
+/// `conv=swab` or `conv=block` command-line arguments. This function
-    bsize: usize,
+/// mutates the `buf` argument in-place. The returned [`ReadStat`]
-) -> std::io::Result<ReadStat> {
+/// indicates how many blocks were read.
+fn read_helper(i: &mut Input, buf: &mut Vec<u8>, bsize: usize) -> std::io::Result<ReadStat> {
     // Local Helper Fns -------------------------------------------------
     fn perform_swab(buf: &mut [u8]) {
         for base in (1..buf.len()).step_by(2) {
@@ -778,6 +885,17 @@ fn is_stdout_redirected_to_seekable_file() -> bool {
     }
 }
 
+/// Decide whether the named file is a named pipe, also known as a FIFO.
+#[cfg(unix)]
+fn is_fifo(filename: &str) -> bool {
+    if let Ok(metadata) = std::fs::metadata(filename) {
+        if metadata.file_type().is_fifo() {
+            return true;
+        }
+    }
+    false
+}
+
 #[uucore::main]
 pub fn uumain(args: impl uucore::Args) -> UResult<()> {
     let args = args.collect_ignore();
@@ -792,40 +910,22 @@ pub fn uumain(args: impl uucore::Args) -> UResult<()> {
             .collect::<Vec<_>>()[..],
     )?;
 
-    match (&settings.infile, &settings.outfile) {
+    let i = match settings.infile {
-        (Some(infile), Some(outfile)) => {
+        #[cfg(unix)]
-            let i = Input::<File>::new(Path::new(&infile), &settings)?;
+        Some(ref infile) if is_fifo(infile) => Input::new_fifo(Path::new(&infile), &settings)?,
-            let o = Output::new_file(Path::new(&outfile), &settings)?;
+        Some(ref infile) => Input::new_file(Path::new(&infile), &settings)?,
-            o.dd_out(i).map_err_context(|| "IO error".to_string())
+        None => Input::new_stdin(&settings)?,
+    };
+    let o = match settings.outfile {
+        #[cfg(unix)]
+        Some(ref outfile) if is_fifo(outfile) => Output::new_fifo(Path::new(&outfile), &settings)?,
+        Some(ref outfile) => Output::new_file(Path::new(&outfile), &settings)?,
+        None if is_stdout_redirected_to_seekable_file() => {
+            Output::new_file(Path::new(&stdout_canonicalized()), &settings)?
         }
-        (None, Some(outfile)) => {
+        None => Output::new_stdout(&settings)?,
-            let i = Input::<io::Stdin>::new(&settings)?;
+    };
-            let o = Output::new_file(Path::new(&outfile), &settings)?;
+    o.dd_out(i).map_err_context(|| "IO error".to_string())
-            o.dd_out(i).map_err_context(|| "IO error".to_string())
-        }
-        (Some(infile), None) => {
-            let i = Input::<File>::new(Path::new(&infile), &settings)?;
-            if is_stdout_redirected_to_seekable_file() {
-                let filename = stdout_canonicalized();
-                let o = Output::new_file(Path::new(&filename), &settings)?;
-                o.dd_out(i).map_err_context(|| "IO error".to_string())
-            } else {
-                let o = Output::new_stdout(&settings)?;
-                o.dd_out(i).map_err_context(|| "IO error".to_string())
-            }
-        }
-        (None, None) => {
-            let i = Input::<io::Stdin>::new(&settings)?;
-            if is_stdout_redirected_to_seekable_file() {
-                let filename = stdout_canonicalized();
-                let o = Output::new_file(Path::new(&filename), &settings)?;
-                o.dd_out(i).map_err_context(|| "IO error".to_string())
-            } else {
-                let o = Output::new_stdout(&settings)?;
-                o.dd_out(i).map_err_context(|| "IO error".to_string())
-            }
-        }
-    }
 }
 
 pub fn uu_app() -> Command {

tests/by-util/test_dd.rs

@@ -5,6 +5,8 @@ use crate::common::util::*;
 use std::fs::{File, OpenOptions};
 use std::io::{BufReader, Read, Write};
 use std::path::PathBuf;
+#[cfg(all(not(windows), not(target_os = "macos")))]
+use std::process::{Command, Stdio};
 #[cfg(not(windows))]
 use std::thread::sleep;
 #[cfg(not(windows))]
@@ -1442,3 +1444,73 @@ fn test_sparse() {
     // number of blocks stored on disk may be zero.
     assert_eq!(at.metadata("infile").len(), at.metadata("outfile").len());
 }
+
+// TODO These FIFO tests should work on macos, but some issue is
+// causing our implementation of dd to wait indefinitely when it
+// shouldn't.
+
+/// Test that a seek on an output FIFO results in a read.
+#[test]
+#[cfg(all(unix, not(target_os = "macos"), not(target_os = "freebsd")))]
+fn test_seek_output_fifo() {
+    let ts = TestScenario::new(util_name!());
+    let at = &ts.fixtures;
+    at.mkfifo("fifo");
+
+    // TODO When `dd` is a bit more advanced, we could use the uutils
+    // version of dd here as well.
+    let child = Command::new("dd")
+        .current_dir(&at.subdir)
+        .args([
+            "count=1",
+            "if=/dev/zero",
+            &format!("of={}", at.plus_as_string("fifo")),
+            "status=noxfer",
+        ])
+        .stderr(Stdio::piped())
+        .spawn()
+        .expect("failed to execute child process");
+
+    ts.ucmd()
+        .args(&["count=0", "seek=1", "of=fifo", "status=noxfer"])
+        .succeeds()
+        .stderr_only("0+0 records in\n0+0 records out\n");
+
+    let output = child.wait_with_output().unwrap();
+    assert!(output.status.success());
+    assert!(output.stdout.is_empty());
+    assert_eq!(&output.stderr, b"1+0 records in\n1+0 records out\n");
+}
+
+/// Test that a skip on an input FIFO results in a read.
+#[test]
+#[cfg(all(unix, not(target_os = "macos"), not(target_os = "freebsd")))]
+fn test_skip_input_fifo() {
+    let ts = TestScenario::new(util_name!());
+    let at = &ts.fixtures;
+    at.mkfifo("fifo");
+
+    // TODO When `dd` is a bit more advanced, we could use the uutils
+    // version of dd here as well.
+    let child = Command::new("dd")
+        .current_dir(&at.subdir)
+        .args([
+            "count=1",
+            "if=/dev/zero",
+            &format!("of={}", at.plus_as_string("fifo")),
+            "status=noxfer",
+        ])
+        .stderr(Stdio::piped())
+        .spawn()
+        .expect("failed to execute child process");
+
+    ts.ucmd()
+        .args(&["count=0", "skip=1", "if=fifo", "status=noxfer"])
+        .succeeds()
+        .stderr_only("0+0 records in\n0+0 records out\n");
+
+    let output = child.wait_with_output().unwrap();
+    assert!(output.status.success());
+    assert!(output.stdout.is_empty());
+    assert_eq!(&output.stderr, b"1+0 records in\n1+0 records out\n");
+}