pipes.rs: don't mix raw write and buffered write

src/uucore/Cargo.toml

@@ -170,7 +170,7 @@ parser-num = ["extendedbigdecimal", "num-traits"]
 parser-size = ["parser-num", "procfs"]
 parser-glob = ["glob"]
 parser = ["parser-num", "parser-size", "parser-glob"]
-pipes = []
+pipes = ["fs"]
 process = ["libc"]
 proc-info = ["tty", "walkdir"]
 quoting-style = ["i18n-common"]

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

@@ -10,7 +10,11 @@ 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::Read, os::fd::AsFd, sync::OnceLock};
+use std::{
+    io::{Read, Write},
+    os::fd::AsFd,
+    sync::OnceLock,
+};
 #[cfg(any(target_os = "linux", target_os = "android"))]
 pub const MAX_ROOTLESS_PIPE_SIZE: usize = 1024 * 1024;
 #[cfg(any(target_os = "linux", target_os = "android"))]
@@ -93,7 +97,7 @@ pub fn might_fuse(source: &impl AsFd) -> bool {
 pub fn splice_unbounded<R, S>(source: &R, dest: &mut S) -> std::io::Result<bool>
 where
     R: Read + AsFd,
-    S: AsFd + std::io::Write,
+    S: AsFd + Write,
 {
     // improve throughput
     // todo: avoid fcntl overhead for small input, but don't fcntl inside of the loop
@@ -119,7 +123,7 @@ where
 pub fn splice_unbounded_broker<R, S>(source: &R, dest: &mut S) -> std::io::Result<bool>
 where
     R: Read + AsFd,
-    S: AsFd + std::io::Write,
+    S: AsFd + Write,
 {
     static PIPE_CACHE: OnceLock<Option<(File, File)>> = OnceLock::new();
     let Some((pipe_rd, pipe_wr)) = PIPE_CACHE.get_or_init(|| pipe().ok()).as_ref() else {
@@ -139,12 +143,12 @@ where
                     // 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
+                    // 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)?;
-                    dest.write_all(&drain)?;
+                    crate::io::RawWriter(&dest).write_all(&drain)?;
                     return Ok(true);
                 }
             }
@@ -159,7 +163,7 @@ where
 #[cfg(any(target_os = "linux", target_os = "android"))]
 pub fn send_n_bytes(
     input: impl Read + AsFd,
-    mut target: impl std::io::Write + AsFd,
+    mut target: impl Write + AsFd,
     n: u64,
 ) -> std::io::Result<u64> {
     static PIPE_CACHE: OnceLock<Option<(File, File)>> = OnceLock::new();
@@ -210,9 +214,10 @@ pub fn send_n_bytes(
                         }
                     } else {
                         debug_assert!(s <= MAX_ROOTLESS_PIPE_SIZE, "unexpected RAM usage");
+                        // drain pipe before fallback to raw write
                         let mut drain = Vec::with_capacity(s);
                         broker_r.take(s as u64).read_to_end(&mut drain)?;
-                        target.write_all(&drain)?;
+                        crate::io::RawWriter(&target).write_all(&drain)?;
                         break true;
                     }
                 }

src/uucore/src/lib/mods/io.rs

@@ -30,11 +30,11 @@ type NativeType = OwnedHandle;
 #[cfg(not(windows))]
 type NativeType = OwnedFd;
 
-// create stdout without buffering
+// create writer without buffering
 #[cfg(any(unix, target_os = "wasi"))]
-pub struct RawWriter(pub io::Stdout);
+pub struct RawWriter<T: AsFd>(pub T);
 #[cfg(any(unix, target_os = "wasi"))]
-impl io::Write for RawWriter {
+impl<T: AsFd> io::Write for RawWriter<T> {
     fn write(&mut self, b: &[u8]) -> io::Result<usize> {
         rustix::io::write(&self.0, b).map_err(Into::into)
     }