refactor: replace rustix::pipe with std::io::pipe

Author: xtqqczze

fuzz/uufuzz/src/lib.rs

@@ -11,11 +11,11 @@ use rand::RngExt;
 use rand::prelude::IndexedRandom;
 use rustix::io::dup;
 use rustix::io::read;
-use rustix::pipe::pipe;
 use rustix::stdio::{dup2_stderr, dup2_stdin, dup2_stdout};
 use std::env::temp_dir;
 use std::ffi::OsString;
 use std::fs::File;
+use std::io::pipe;
 use std::io::{Seek, SeekFrom, Write};
 use std::process::{Command, Stdio};
 use std::sync::atomic::Ordering;
@@ -92,7 +92,7 @@ where
     };
 
     println!("Running test {:?}", &args[0..]);
-    let (read_pipe_stdout, write_pipe_stdout) = match pipe() {
+    let (stdout_pipe_reader, stdout_pipe_writer) = match pipe() {
         Ok(fds) => fds,
         Err(_) => {
             return CommandResult {
@@ -102,7 +102,7 @@ where
             };
         }
     };
-    let (read_pipe_stderr, write_pipe_stderr) = match pipe() {
+    let (stderr_pipe_reader, stderr_pipe_writer) = match pipe() {
         Ok(fds) => fds,
         Err(_) => {
             return CommandResult {
@@ -114,7 +114,7 @@ where
     };
 
     // Redirect stdout and stderr to their respective pipes
-    if dup2_stdout(&write_pipe_stdout).is_err() || dup2_stderr(&write_pipe_stderr).is_err() {
+    if dup2_stdout(&stdout_pipe_writer).is_err() || dup2_stderr(&stderr_pipe_writer).is_err() {
         return CommandResult {
             stdout: "".to_string(),
             stderr: "Failed to redirect STDOUT_FILENO or STDERR_FILENO".to_string(),
@@ -156,8 +156,8 @@ where
     };
 
     let (uumain_exit_status, captured_stdout, captured_stderr) = thread::scope(|s| {
-        let out = s.spawn(|| read_from_fd(read_pipe_stdout));
-        let err = s.spawn(|| read_from_fd(read_pipe_stderr));
+        let out = s.spawn(|| read_from_fd(stdout_pipe_reader));
+        let err = s.spawn(|| read_from_fd(stderr_pipe_reader));
         #[allow(clippy::unnecessary_to_owned)]
         // TODO: clippy wants us to use args.iter().cloned() ?
         let status = uumain_function(args.to_owned().into_iter());
@@ -167,8 +167,8 @@ where
         io::stdout().flush().unwrap();
         io::stderr().flush().unwrap();
         // Drop write ends to close them, allowing readers to get EOF
-        drop(write_pipe_stdout);
-        drop(write_pipe_stderr);
+        drop(stdout_pipe_writer);
+        drop(stderr_pipe_writer);
         // Restore stdout/stderr
         let _ = dup2_stdout(&original_stdout_fd_owned);
         let _ = dup2_stderr(&original_stderr_fd_owned);

src/uu/wc/src/count_fast.rs

@@ -55,13 +55,14 @@ fn count_bytes_using_splice(fd: &impl AsFd) -> Result<usize, usize> {
         }
     } else {
         // input is not pipe. needs broker to use splice() with additional cost
-        let (pipe_rd, pipe_wr) = pipe::<false>(MAX_ROOTLESS_PIPE_SIZE).map_err(|_| 0_usize)?;
+        let (pipe_reader, pipe_writer) =
+            pipe::<false>(MAX_ROOTLESS_PIPE_SIZE).map_err(|_| 0_usize)?;
         loop {
-            match splice(fd, &pipe_wr, MAX_ROOTLESS_PIPE_SIZE) {
+            match splice(fd, &pipe_writer, MAX_ROOTLESS_PIPE_SIZE) {
                 Ok(0) => return Ok(byte_count),
                 Ok(res) => {
                     byte_count += res;
-                    splice_exact(&pipe_rd, &null_file, res).map_err(|_| byte_count)?;
+                    splice_exact(&pipe_reader, &null_file, res).map_err(|_| byte_count)?;
                 }
                 Err(_) => return Err(byte_count),
             }

src/uu/yes/src/yes.rs

@@ -117,18 +117,19 @@ pub fn exec(mut bytes: Vec<u8>) -> io::Result<()> {
     // improve throughput
     let _ = rustix::pipe::fcntl_setpipe_size(&stdout, MAX_ROOTLESS_PIPE_SIZE);
     // don't show any error from fast-path and fallback to write for proper message
-    if let Ok((p_read, mut p_write)) = pipe::<true>(MAX_ROOTLESS_PIPE_SIZE)
-        && p_write.write_all(bytes).is_ok()
+    if let Ok((source_reader, mut source_writer)) = pipe::<true>(MAX_ROOTLESS_PIPE_SIZE)
+        && source_writer.write_all(bytes).is_ok()
     {
-        if aligned && tee(&p_read, &stdout, MAX_ROOTLESS_PIPE_SIZE).is_ok() {
-            while let Ok(1..) = tee(&p_read, &stdout, MAX_ROOTLESS_PIPE_SIZE) {}
-        } else if let Ok((broker_read, broker_write)) = pipe::<true>(MAX_ROOTLESS_PIPE_SIZE) {
+        if aligned && tee(&source_reader, &stdout, MAX_ROOTLESS_PIPE_SIZE).is_ok() {
+            while let Ok(1..) = tee(&source_reader, &stdout, MAX_ROOTLESS_PIPE_SIZE) {}
+        } else if let Ok((broker_reader, broker_writer)) = pipe::<true>(MAX_ROOTLESS_PIPE_SIZE) {
             // tee() cannot control offset and write to non-pipe
-            'hybrid: while let Ok(mut remain) = tee(&p_read, &broker_write, MAX_ROOTLESS_PIPE_SIZE)
+            'hybrid: while let Ok(mut remain) =
+                tee(&source_reader, &broker_writer, MAX_ROOTLESS_PIPE_SIZE)
             {
                 debug_assert!(remain == bytes.len(), "splice() should cleanup pipe");
                 while remain > 0 {
-                    if let Ok(s) = splice(&broker_read, &stdout, remain) {
+                    if let Ok(s) = splice(&broker_reader, &stdout, remain) {
                         remain -= s;
                     } else {
                         // avoid output breakage with reduced remain even if it would not happen

src/uucore/src/lib/features/buf_copy.rs

@@ -51,14 +51,12 @@ mod tests {
     fn test_copy_stream() {
         let mut dest_file = new_temp_file();
 
-        let (pipe_read, pipe_write) = rustix::pipe::pipe().unwrap();
-        let mut pipe_read: File = pipe_read.into();
-        let mut pipe_write: File = pipe_write.into();
+        let (mut pipe_reader, mut pipe_writer) = std::io::pipe().unwrap();
         let data = b"Hello, world!";
         let thread = thread::spawn(move || {
-            pipe_write.write_all(data).unwrap();
+            pipe_writer.write_all(data).unwrap();
         });
-        copy_stream(&mut pipe_read, &mut dest_file).unwrap();
+        copy_stream(&mut pipe_reader, &mut dest_file).unwrap();
         thread.join().unwrap();
 
         // We would have been at the end already, so seek again to the start.

src/uucore/src/lib/features/pipes.rs

@@ -10,6 +10,8 @@ use rustix::pipe::{SpliceFlags, fcntl_setpipe_size};
 #[cfg(any(target_os = "linux", target_os = "android"))]
 use std::fs::File;
 #[cfg(any(target_os = "linux", target_os = "android"))]
+use std::io::{PipeReader, PipeWriter};
+#[cfg(any(target_os = "linux", target_os = "android"))]
 use std::{
     io::{Read, Write},
     os::fd::AsFd,
@@ -27,17 +29,17 @@ const KERNEL_DEFAULT_PIPE_SIZE: usize = 64 * 1024;
 /// used for resolving the limitation for splice: one of a input or output should be pipe
 #[inline]
 #[cfg(any(target_os = "linux", target_os = "android"))]
-pub fn pipe<const SIZE_REQUIRED: bool>(s: usize) -> std::io::Result<(File, File)> {
-    let (read, write) = rustix::pipe::pipe()?;
+pub fn pipe<const SIZE_REQUIRED: bool>(s: usize) -> std::io::Result<(PipeReader, PipeWriter)> {
+    let (pipe_reader, pipe_writer) = std::io::pipe()?;
     // guard unnecessary syscall
     if s > KERNEL_DEFAULT_PIPE_SIZE {
-        let r = fcntl_setpipe_size(&read, s);
+        let r = fcntl_setpipe_size(&pipe_reader, s);
         if SIZE_REQUIRED {
             r?;
         }
     }
 
-    Ok((File::from(read), File::from(write)))
+    Ok((pipe_reader, pipe_writer))
 }
 
 /// Less noisy wrapper around [`rustix::pipe::splice`].
@@ -116,8 +118,8 @@ where
     R: Read + AsFd,
     S: AsFd,
 {
-    static PIPE_CACHE: OnceLock<Option<(File, File)>> = OnceLock::new();
-    let Some((pipe_rd, pipe_wr)) = PIPE_CACHE
+    static PIPE_CACHE: OnceLock<Option<(PipeReader, PipeWriter)>> = OnceLock::new();
+    let Some((pipe_reader, pipe_writer)) = PIPE_CACHE
         .get_or_init(|| pipe::<false>(MAX_ROOTLESS_PIPE_SIZE).ok())
         .as_ref()
     else {
@@ -130,18 +132,18 @@ where
     let _ = fcntl_setpipe_size(&mut *dest, MAX_ROOTLESS_PIPE_SIZE);
 
     loop {
-        match splice(&source, &pipe_wr, MAX_ROOTLESS_PIPE_SIZE) {
+        match splice(&source, &pipe_writer, MAX_ROOTLESS_PIPE_SIZE) {
             Ok(0) => return Ok(false),
             Ok(n) => {
-                if splice_exact(&pipe_rd, dest, n).is_err() {
+                if splice_exact(&pipe_reader, dest, 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 unbuffered read/write. Then
                     // we tell the caller to fall back.
                     debug_assert!(n <= MAX_ROOTLESS_PIPE_SIZE, "unexpected RAM usage");
                     let mut drain = Vec::with_capacity(n);
-                    pipe_rd.take(n as u64).read_to_end(&mut drain)?;
+                    pipe_reader.take(n as u64).read_to_end(&mut drain)?;
                     crate::io::RawWriter(&dest).write_all(&drain)?;
                     return Ok(true);
                 }
@@ -160,7 +162,7 @@ pub fn send_n_bytes(
     mut target: impl Write + AsFd,
     n: u64,
 ) -> std::io::Result<u64> {
-    static PIPE_CACHE: OnceLock<Option<(File, File)>> = OnceLock::new();
+    static PIPE_CACHE: OnceLock<Option<(PipeReader, PipeWriter)>> = OnceLock::new();
     let pipe_size = MAX_ROOTLESS_PIPE_SIZE.min(n as usize);
     let mut n = n;
     let mut bytes_written: u64 = 0;

tests/by-util/test_cat.rs

@@ -23,15 +23,14 @@ use uutests::util_name;
 #[test]
 fn test_cat_broken_pipe_nonzero_and_message() {
     use uutests::new_ucmd;
-    let (read, write) = rustix::pipe::pipe().expect("Failed to create pipe");
+    let (pipe_reader, pipe_writer) = std::io::pipe().expect("Failed to create pipe");
     // Close the read end to simulate a broken pipe on stdout
-    drop(read);
-    let write: File = write.into();
+    drop(pipe_reader);
     let content = (0..10000).map(|_| "x").collect::<String>();
 
     // On Unix, SIGPIPE should lead to a non-zero exit; ensure process exits and fails
     new_ucmd!()
-        .set_stdout(write)
+        .set_stdout(pipe_writer)
         .pipe_in(content.as_bytes())
         .fails();
 }

tests/by-util/test_comm.rs

@@ -691,10 +691,8 @@ fn test_comm_anonymous_pipes() {
     let scene = TestScenario::new(util_name!());
 
     // Open two anonymous pipes
-    let (comm1_reader, comm1_writer) = rustix::pipe::pipe().unwrap();
-    let mut comm1_writer: std::fs::File = comm1_writer.into();
-    let (comm2_reader, comm2_writer) = rustix::pipe::pipe().unwrap();
-    let mut comm2_writer: std::fs::File = comm2_writer.into();
+    let (comm1_reader, mut comm1_writer) = std::io::pipe().unwrap();
+    let (comm2_reader, mut comm2_writer) = std::io::pipe().unwrap();
 
     // comm reads the data in chunks
     // make content large enough, so that at least two chunks are read

tests/by-util/test_paste.rs

@@ -450,12 +450,12 @@ fn test_paste_non_utf8_paths() {
 }
 
 #[cfg(target_os = "linux")]
-fn make_broken_pipe() -> std::fs::File {
-    let (read, write) = rustix::pipe::pipe().expect("Failed to create pipe");
+fn make_broken_pipe() -> std::io::PipeWriter {
+    let (reader, writer) = std::io::pipe().expect("Failed to create pipe");
     // Drop the read end so writes fail with EPIPE.
-    drop(read);
+    drop(reader);
 
-    write.into()
+    writer
 }
 
 #[test]

tests/by-util/test_tee.rs

@@ -256,26 +256,25 @@ mod linux_only {
     use uutests::util::{AtPath, CmdResult, UCommand};
 
     use std::fmt::Write;
-    use std::fs::File;
     use std::process::Stdio;
     use std::time::Duration;
     use uutests::at_and_ucmd;
     use uutests::new_ucmd;
 
-    fn make_broken_pipe() -> File {
-        let (read, write) = rustix::pipe::pipe().expect("Failed to create pipe");
+    fn make_broken_pipe() -> std::io::PipeWriter {
+        let (reader, writer) = std::io::pipe().expect("Failed to create pipe");
         // Drop the read end of the pipe
-        drop(read);
-        // Make the write end of the pipe into a Rust File
-        write.into()
+        drop(reader);
+        // Return the write end of the pipe
+        writer
     }
 
-    fn make_hanging_read() -> File {
-        let (read, write) = rustix::pipe::pipe().expect("Failed to create pipe");
+    fn make_hanging_read() -> std::io::PipeReader {
+        let (reader, writer) = std::io::pipe().expect("Failed to create pipe");
         // PURPOSELY leak the write end of the pipe, so the read end hangs.
-        std::mem::forget(write);
+        std::mem::forget(writer);
         // Return the read end of the pipe
-        read.into()
+        reader
     }
 
     fn run_tee(proc: &mut UCommand) -> (String, CmdResult) {
@@ -726,14 +725,13 @@ fn test_output_error_flag_without_value_defaults_warn_nopipe() {
 #[cfg(all(unix, not(target_os = "freebsd")))]
 #[test]
 fn test_output_error_presence_only_broken_pipe_unix() {
-    let (read, write) = rustix::pipe::pipe().expect("Failed to create pipe");
+    let (reader, writer) = std::io::pipe().expect("Failed to create pipe");
     // Close the read end to simulate a broken pipe on stdout
-    drop(read);
-    let write: std::fs::File = write.into();
+    drop(reader);
     let content = (0..10_000).map(|_| "x").collect::<String>();
     let result = new_ucmd!()
         .arg("--output-error") // presence-only flag
-        .set_stdout(write)
+        .set_stdout(writer)
         .pipe_in(content.as_bytes())
         .run();
 
@@ -745,14 +743,13 @@ fn test_output_error_presence_only_broken_pipe_unix() {
 #[cfg(all(unix, not(target_os = "freebsd")))]
 #[test]
 fn test_broken_pipe_early_termination_stdout_only() {
-    let (read, write) = rustix::pipe::pipe().expect("Failed to create pipe");
+    let (reader, writer) = std::io::pipe().expect("Failed to create pipe");
     // Create a broken stdout
-    drop(read);
-    let write: std::fs::File = write.into();
+    drop(reader);
     let content = (0..10_000).map(|_| "x").collect::<String>();
     let mut proc = new_ucmd!();
     let result = proc
-        .set_stdout(write)
+        .set_stdout(writer)
         .ignore_stdin_write_error()
         .pipe_in(content.as_bytes())
         .run();