From 7dd1a40ce0dddc27cabe88c476010631bafed606 Mon Sep 17 00:00:00 2001 From: oech3 <79379754+oech3@users.noreply.github.com> Date: Tue, 21 Apr 2026 20:19:19 +0900 Subject: [PATCH] uucore: dedup code of unbounded splice with broker --- src/uu/cat/src/splice.rs | 32 +++---------- src/uucore/src/lib/features/buf_copy/linux.rs | 48 ++----------------- src/uucore/src/lib/features/pipes.rs | 43 +++++++++++++++++ 3 files changed, 52 insertions(+), 71 deletions(-) diff --git a/src/uu/cat/src/splice.rs b/src/uu/cat/src/splice.rs index 8096166f063..3b75b315b16 100644 --- a/src/uu/cat/src/splice.rs +++ b/src/uu/cat/src/splice.rs @@ -4,10 +4,10 @@ // file that was distributed with this source code. use super::{CatResult, FdReadable, InputHandle}; -use std::io::{Read, Write}; +use std::io::Write; use std::os::{fd::AsFd, unix::io::AsRawFd}; -use uucore::pipes::{MAX_ROOTLESS_PIPE_SIZE, might_fuse, pipe, splice, splice_exact}; +use uucore::pipes::{MAX_ROOTLESS_PIPE_SIZE, might_fuse, splice}; /// This function is called from `write_fast()` on Linux and Android. The /// function `splice()` is used to move data between two file descriptors @@ -21,8 +21,6 @@ pub(super) fn write_fast_using_splice( handle: &InputHandle, write_fd: &mut S, ) -> CatResult { - use std::{fs::File, sync::OnceLock}; - static PIPE_CACHE: OnceLock> = OnceLock::new(); if splice(&handle.reader, &write_fd, MAX_ROOTLESS_PIPE_SIZE).is_ok() { // fcntl improves throughput // todo: avoid fcntl overhead for small input, but don't fcntl inside of the loop @@ -34,28 +32,10 @@ pub(super) fn write_fast_using_splice( Err(_) => return Ok(true), } } - } else if let Some((pipe_rd, pipe_wr)) = PIPE_CACHE.get_or_init(|| pipe().ok()).as_ref() { - // both of in/output are not pipe. needs broker to use splice() with additional costs - loop { - match splice(&handle.reader, &pipe_wr, MAX_ROOTLESS_PIPE_SIZE) { - Ok(0) => return Ok(might_fuse(&handle.reader)), - Ok(n) => { - 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. - let mut drain = Vec::with_capacity(n); // bounded by pipe size - pipe_rd.take(n as u64).read_to_end(&mut drain)?; - write_fd.write_all(&drain)?; - return Ok(true); - } - } - Err(_) => return Ok(true), - } - } } else { - Ok(true) + Ok( + uucore::pipes::splice_unbounded_broker(&handle.reader, write_fd)? + || might_fuse(&handle.reader), + ) } } diff --git a/src/uucore/src/lib/features/buf_copy/linux.rs b/src/uucore/src/lib/features/buf_copy/linux.rs index 02fea00cd4c..932d9788ec4 100644 --- a/src/uucore/src/lib/features/buf_copy/linux.rs +++ b/src/uucore/src/lib/features/buf_copy/linux.rs @@ -5,10 +5,7 @@ //! Buffer-based copying implementation for Linux and Android. -use crate::{ - error::UResult, - pipes::{MAX_ROOTLESS_PIPE_SIZE, pipe, splice, splice_exact}, -}; +use crate::error::UResult; /// Buffer-based copying utilities for unix (excluding Linux). use std::{ @@ -53,7 +50,8 @@ where { // 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(src, dest)? { + // todo: bypass broker pipe this if input or output is pipe. We use this mostly for stream. + if !crate::pipes::splice_unbounded_broker(src, dest)? { return Ok(()); } @@ -68,43 +66,3 @@ where dest.flush()?; Ok(()) } - -/// Write from source `handle` into destination `write_fd` using Linux-specific -/// `splice` system call. -/// -/// # Arguments -/// - `source` - source handle -/// - `dest` - destination handle -#[inline] -pub(crate) fn splice_write(source: &R, dest: &mut S) -> UResult -where - R: Read + AsFd + AsRawFd, - S: AsRawFd + AsFd + Write, -{ - let (pipe_rd, pipe_wr) = pipe()?; // todo: bypass this if input or output is pipe. We use this mostly for stream. - // improve throughput - // no need to increase pipe size of input fd since - // - sender with splice probably increased size already - // - sender without splice is bottleneck - let _ = rustix::pipe::fcntl_setpipe_size(&mut *dest, MAX_ROOTLESS_PIPE_SIZE); - - loop { - match splice(&source, &pipe_wr, MAX_ROOTLESS_PIPE_SIZE) { - Ok(0) => return Ok(false), - Ok(n) => { - if splice_exact(&pipe_rd, 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 normal read/write. Then - // we tell the caller to fall back. - let mut drain = Vec::with_capacity(n); // bounded by pipe size - pipe_rd.take(n as u64).read_to_end(&mut drain)?; - dest.write_all(&drain)?; - return Ok(true); - } - } - Err(_) => return Ok(true), - } - } -} diff --git a/src/uucore/src/lib/features/pipes.rs b/src/uucore/src/lib/features/pipes.rs index 05a15009b7c..292cf4c4162 100644 --- a/src/uucore/src/lib/features/pipes.rs +++ b/src/uucore/src/lib/features/pipes.rs @@ -86,6 +86,48 @@ pub fn might_fuse(source: &impl AsFd) -> bool { rustix::fs::fstatfs(source).map_or(true, |stats| stats.f_type == 0x6573_5546) // FUSE magic number, too many platform specific clippy warning with const } +/// force-splice source to dest even both of them are not pipe +/// return true if we need read/write fallback +/// +/// This should not be used if one of them are pipe to save resources +#[inline] +#[cfg(any(target_os = "linux", target_os = "android"))] +pub fn splice_unbounded_broker(source: &R, dest: &mut S) -> std::io::Result +where + R: Read + AsFd, + S: AsFd + std::io::Write, +{ + static PIPE_CACHE: OnceLock> = OnceLock::new(); + let Some((pipe_rd, pipe_wr)) = PIPE_CACHE.get_or_init(|| pipe().ok()).as_ref() else { + return Ok(true); + }; + // improve throughput + // no need to increase pipe size of input fd since + // - sender with splice probably increased size already + // - sender without splice is bottleneck + let _ = fcntl_setpipe_size(&mut *dest, MAX_ROOTLESS_PIPE_SIZE); + + loop { + match splice(&source, &pipe_wr, MAX_ROOTLESS_PIPE_SIZE) { + Ok(0) => return Ok(false), + Ok(n) => { + if splice_exact(&pipe_rd, 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 normal read/write. Then + // we tell the caller to fall back. + let mut drain = Vec::with_capacity(n); // bounded by pipe size + pipe_rd.take(n as u64).read_to_end(&mut drain)?; + dest.write_all(&drain)?; + return Ok(true); + } + } + Err(_) => return Ok(true), + } + } +} + /// splice `n` bytes with safe read/write fallback /// return actually sent bytes #[inline] @@ -129,6 +171,7 @@ pub fn send_n_bytes( .get_or_init(|| pipe_with_size(pipe_size).ok()) .as_ref() { + // todo: create fn splice_bounded_broker loop { match splice(&input, &broker_w, n as usize) { Ok(0) => break might_fuse(&input),