diff --git a/Cargo.lock b/Cargo.lock index 64dba2c9..53053108 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -519,6 +519,7 @@ dependencies = [ "itertools 0.15.0", "nix 0.31.3", "onig", + "rayon", "regex", "rstest", "tempfile", diff --git a/Cargo.toml b/Cargo.toml index cb172f36..19b79844 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -13,6 +13,7 @@ authors = ["uutils developers"] argmax = "0.4.0" chrono = "0.4.45" clap = { version = "4.6", features = ["env"] } +rayon = "1" faccess = "0.2.4" nix = { version = "0.31", features = ["fs", "user"] } onig = { version = "6.5", default-features = false } diff --git a/benches/xargs_bench.rs b/benches/xargs_bench.rs index 2be2a057..23c48829 100644 --- a/benches/xargs_bench.rs +++ b/benches/xargs_bench.rs @@ -9,6 +9,9 @@ //! can feed a fixed corpus without touching the process's stdin, and the //! command is `true` so the measurement is dominated by `xargs`'s own work //! (reading, splitting and batching arguments) rather than the child. +//! +//! Both the serial and the parallel (`-P`) paths are covered so the rayon-based +//! dispatch introduced for `xargs -P` is measured alongside the serial one. use std::path::PathBuf; @@ -47,6 +50,13 @@ fn bench_e2e(c: &mut Criterion) { let ws_path = ws.to_str().unwrap(); let nul_path = nul.to_str().unwrap(); + // A smaller corpus for the parallel benchmarks: each batch spawns a real + // `true` child, so keeping the token (and therefore batch) count modest + // avoids thousands of spawns per iteration while still splitting the work + // into enough batches to feed the parallel bridge. + let ws_small = build_input(400, " ", "ws_small"); + let ws_small_path = ws_small.to_str().unwrap(); + let mut group = c.benchmark_group("xargs"); // Default whitespace splitting, single batch (fits one command line). @@ -67,8 +77,58 @@ fn bench_e2e(c: &mut Criterion) { b.iter(|| run(black_box(&["-a", ws_path, "-s", "4096", "true"]))); }); + // Parallel execution (`-P`) drives the rayon-based path in `process_input`. + // `-n 10` splits the small corpus into a handful of batches that are + // dispatched through the parallel bridge, so the measurement captures the + // parallel dispatch and batching overhead. Two thread counts are covered: a + // low degree of parallelism (`-P2`) and a higher one (`-P8`) to exercise the + // scheduler under more contention. + group.bench_function("parallel_p2", |b| { + b.iter(|| { + run(black_box(&[ + "-a", + ws_small_path, + "-n", + "10", + "-P", + "2", + "true", + ])); + }); + }); + group.bench_function("parallel_p8", |b| { + b.iter(|| { + run(black_box(&[ + "-a", + ws_small_path, + "-n", + "10", + "-P", + "8", + "true", + ])); + }); + }); + // Parallel path with larger batches (`-n 50`): fewer, bigger commands so the + // balance shifts back towards argument assembly while still going through + // the parallel bridge. + group.bench_function("parallel_p8_batched_n", |b| { + b.iter(|| { + run(black_box(&[ + "-a", + ws_small_path, + "-n", + "50", + "-P", + "8", + "true", + ])); + }); + }); + group.finish(); + let _ = std::fs::remove_file(&ws_small); let _ = std::fs::remove_file(&ws); let _ = std::fs::remove_file(&nul); } diff --git a/src/xargs/mod.rs b/src/xargs/mod.rs index ff833335..97efc6d0 100644 --- a/src/xargs/mod.rs +++ b/src/xargs/mod.rs @@ -12,9 +12,15 @@ use std::{ fs, io::{self, BufRead, BufReader, Read}, process::{Command, Stdio}, + sync::atomic::{AtomicU64, Ordering}, }; +static ACTIVE_PROCS: AtomicU64 = AtomicU64::new(0); + use clap::{crate_version, error::ErrorKind, Arg, ArgAction}; +use rayon::iter::{ + FromParallelIterator, IntoParallelIterator, ParallelBridge as _, ParallelIterator as _, +}; mod options { pub const COMMAND: &str = "COMMAND"; @@ -79,14 +85,14 @@ trait CommandSizeLimiter { arg: Argument, cursor: LimiterCursor<'_>, ) -> Result; - fn dyn_clone(&self) -> Box; + fn dyn_clone(&self) -> Box; } /// A pointer to the next limiter. A limiter should *always* call the cursor's /// `try_next` *before* updating its own state, to ensure that all other limiters /// are okay with the argument first. struct LimiterCursor<'collection> { - limiters: &'collection mut [Box], + limiters: &'collection mut [Box], } impl LimiterCursor<'_> { @@ -106,7 +112,7 @@ impl LimiterCursor<'_> { } struct LimiterCollection { - limiters: Vec>, + limiters: Vec>, } impl LimiterCollection { @@ -114,7 +120,7 @@ impl LimiterCollection { Self { limiters: vec![] } } - fn add(&mut self, limiter: impl CommandSizeLimiter + 'static) { + fn add(&mut self, limiter: impl CommandSizeLimiter + Send + Sync + 'static) { self.limiters.push(Box::new(limiter)); } @@ -267,7 +273,7 @@ impl CommandSizeLimiter for MaxCharsCommandSizeLimiter { } } - fn dyn_clone(&self) -> Box { + fn dyn_clone(&self) -> Box { Box::new(self.clone()) } } @@ -307,7 +313,7 @@ impl CommandSizeLimiter for MaxArgsCommandSizeLimiter { } } - fn dyn_clone(&self) -> Box { + fn dyn_clone(&self) -> Box { Box::new(self.clone()) } } @@ -351,21 +357,44 @@ impl CommandSizeLimiter for MaxLinesCommandSizeLimiter { } } - fn dyn_clone(&self) -> Box { + fn dyn_clone(&self) -> Box { Box::new(self.clone()) } } +#[derive(Default)] enum CommandResult { + #[default] Success, Failure, } -impl CommandResult { - fn combine(&mut self, other: Self) { - if matches!(*self, Self::Success) { - *self = other; - } +impl FromIterator for CommandResult { + fn from_iter>(iter: I) -> Self { + iter.into_iter() + .fold(None, |acc, item| { + acc.or_else(|| (!matches!(item, Self::Success)).then_some(item)) + }) + .unwrap_or_default() + } +} + +impl FromParallelIterator for CommandResult { + fn from_par_iter(par_iter: I) -> Self + where + I: IntoParallelIterator, + { + par_iter + .into_par_iter() + .fold( + || None, + |acc, item| acc.or_else(|| (!matches!(item, Self::Success)).then_some(item)), + ) + .collect_vec_list() + .into_iter() + .flatten() + .flatten() + .collect() } } @@ -545,6 +574,20 @@ impl CommandBuilder<'_> { } } } + + fn execute_parallel(self, capacity: u64) -> Result { + loop { + let current = ACTIVE_PROCS.fetch_add(1, Ordering::SeqCst); + if current < capacity { + break; + } + ACTIVE_PROCS.fetch_sub(1, Ordering::SeqCst); + std::hint::spin_loop(); + } + let result = self.execute(); + ACTIVE_PROCS.fetch_sub(1, Ordering::SeqCst); + result + } } trait ArgumentReader { @@ -701,13 +744,13 @@ where } struct EofArgumentReader { - reader: Box, + reader: Box, eof_delimiter: OsString, eof_found: bool, } impl EofArgumentReader { - fn new(reader: Box, eof_delimiter: &String) -> Self { + fn new(reader: Box, eof_delimiter: &String) -> Self { Self { reader, eof_delimiter: eof_delimiter.into(), @@ -783,6 +826,7 @@ struct InputProcessOptions { max_args: Option, max_lines: Option, no_run_if_empty: bool, + max_procs: std::num::NonZero, } impl InputProcessOptions { @@ -791,51 +835,119 @@ impl InputProcessOptions { max_args: Option, max_lines: Option, no_run_if_empty: bool, + max_procs: std::num::NonZero, ) -> Self { Self { exit_if_pass_char_limit, max_args, max_lines, no_run_if_empty, + max_procs, + } + } +} + +struct CommandBatchIter<'a> { + builder_options: &'a CommandBuilderOptions, + args: Box, + options: &'a InputProcessOptions, + current_builder: CommandBuilder<'a>, + have_pending_command: bool, + done: bool, +} + +impl<'a> CommandBatchIter<'a> { + fn new( + builder_options: &'a CommandBuilderOptions, + args: Box, + options: &'a InputProcessOptions, + ) -> Self { + Self { + current_builder: CommandBuilder::new(builder_options), + builder_options, + args, + options, + have_pending_command: false, + done: false, + } + } +} + +impl<'a> Iterator for CommandBatchIter<'a> { + type Item = Result, XargsError>; + + fn next(&mut self) -> Option { + if self.done { + return None; + } + loop { + match self.args.next() { + Err(e) => { + self.done = true; + return Some(Err(XargsError::from(e))); + } + Ok(None) => { + self.done = true; + if !self.options.no_run_if_empty || self.have_pending_command { + return Some(Ok(std::mem::replace( + &mut self.current_builder, + CommandBuilder::new(self.builder_options), + ))); + } + return None; + } + Ok(Some(arg)) => match self.current_builder.add_arg(arg) { + Ok(()) => { + self.have_pending_command = true; + } + Err(ExhaustedCommandSpace { arg, out_of_chars }) => { + if out_of_chars + && self.options.exit_if_pass_char_limit + && (self.options.max_args.is_some() || self.options.max_lines.is_some()) + { + self.done = true; + return Some(Err(XargsError::ArgumentTooLarge)); + } + let old = std::mem::replace( + &mut self.current_builder, + CommandBuilder::new(self.builder_options), + ); + let batch = self.have_pending_command.then_some(old); + if let Err(ExhaustedCommandSpace { .. }) = self.current_builder.add_arg(arg) + { + self.done = true; + return Some(Err(XargsError::ArgumentTooLarge)); + } + self.have_pending_command = true; + if let Some(ready) = batch { + return Some(Ok(ready)); + } + } + }, + } } } } fn process_input( builder_options: &CommandBuilderOptions, - mut args: Box, + args: Box, options: &InputProcessOptions, ) -> Result { - let mut current_builder = CommandBuilder::new(builder_options); - let mut have_pending_command = false; - let mut result = CommandResult::Success; - - while let Some(arg) = args.next()? { - if let Err(ExhaustedCommandSpace { arg, out_of_chars }) = current_builder.add_arg(arg) { - if out_of_chars - && options.exit_if_pass_char_limit - && (options.max_args.is_some() || options.max_lines.is_some()) - { - return Err(XargsError::ArgumentTooLarge); - } - if have_pending_command { - result.combine(current_builder.execute()?); - } + let batches = CommandBatchIter::new(builder_options, args, options); - current_builder = CommandBuilder::new(builder_options); - if let Err(ExhaustedCommandSpace { .. }) = current_builder.add_arg(arg) { - return Err(XargsError::ArgumentTooLarge); - } - } + let capacity = options.max_procs.get(); - have_pending_command = true; - } - - if !options.no_run_if_empty || have_pending_command { - result.combine(current_builder.execute()?); + if capacity == 1 { + batches + .map(|batch| batch?.execute().map_err(XargsError::from)) + .collect() + } else { + batches + .par_bridge() + .map(|batch| batch?.execute_parallel(capacity).map_err(XargsError::from)) + .collect() } - - Ok(result) } fn parse_delimiter(s: &str) -> Result { @@ -1017,7 +1129,7 @@ fn do_xargs(args: &[&str]) -> Result { Arg::new(options::MAX_PROCS) .short('P') .long(options::MAX_PROCS) - .help("Run up to this many commands in parallel [NOT IMPLEMENTED]") + .help("Run up to this many commands in parallel") .value_parser(clap::value_parser!(usize)), ) .arg( @@ -1183,13 +1295,13 @@ fn do_xargs(args: &[&str]) -> Result { builder_options.verbose = options.verbose; builder_options.close_stdin = options.arg_file.is_none(); - let args_file: Box = if let Some(path) = &options.arg_file { + let args_file: Box = if let Some(path) = &options.arg_file { Box::new(fs::File::open(path).map_err(|e| format!("Failed to open {path}: {e}"))?) } else { Box::new(io::stdin()) }; - let mut args: Box = if let Some(delimiter) = options.delimiter { + let mut args: Box = if let Some(delimiter) = options.delimiter { Box::new(ByteDelimitedArgumentReader::new(args_file, delimiter)) } else { Box::new(WhitespaceDelimitedArgumentReader::new(args_file)) @@ -1199,6 +1311,14 @@ fn do_xargs(args: &[&str]) -> Result { args = Box::new(EofArgumentReader::new(args, &eof_delimiter)); } + let max_procs = + matches + .get_one::(options::MAX_PROCS) + .map_or(std::num::NonZero::::MIN, |&n| { + std::num::NonZero::new(n.try_into().expect("max-procs value exceeds u64 range")) + .expect("max-procs must be non-zero") + }); + let result = process_input( &builder_options, args, @@ -1207,6 +1327,7 @@ fn do_xargs(args: &[&str]) -> Result { options.max_args, options.max_lines, options.no_run_if_empty || options.replace.is_some(), + max_procs, ), )?; Ok(result) @@ -1274,7 +1395,7 @@ mod tests { }) } - fn dyn_clone(&self) -> Box { + fn dyn_clone(&self) -> Box { Box::new(self.clone()) } } @@ -1364,7 +1485,8 @@ mod tests { #[test] fn test_chars_limiter_asks_cursor() { - let mut rejects: [Box; 1] = [Box::new(AlwaysRejectLimiter)]; + let mut rejects: [Box; 1] = + [Box::new(AlwaysRejectLimiter)]; let reject_cursor = LimiterCursor { limiters: &mut rejects, }; @@ -1402,7 +1524,8 @@ mod tests { #[test] fn test_args_limiter_asks_cursor() { - let mut rejects: [Box; 1] = [Box::new(AlwaysRejectLimiter)]; + let mut rejects: [Box; 1] = + [Box::new(AlwaysRejectLimiter)]; let reject_cursor = LimiterCursor { limiters: &mut rejects, }; @@ -1448,7 +1571,8 @@ mod tests { #[test] fn test_lines_limiter_asks_cursor() { - let mut rejects: [Box; 1] = [Box::new(AlwaysRejectLimiter)]; + let mut rejects: [Box; 1] = + [Box::new(AlwaysRejectLimiter)]; let reject_cursor = LimiterCursor { limiters: &mut rejects, }; diff --git a/tests/test_xargs.rs b/tests/test_xargs.rs index 997c7e85..cf258f14 100644 --- a/tests/test_xargs.rs +++ b/tests/test_xargs.rs @@ -25,6 +25,24 @@ fn xargs_basics() { .stdout_only("abc def ghi i j \"k\n"); } +#[test] +fn xargs_parallel_one() { + ucmd() + .args(&["-P1"]) + .pipe_in("abc\ndef g\\hi 'i j \"k'") + .succeeds() + .stdout_only("abc def ghi i j \"k\n"); +} + +#[test] +fn xargs_parallel_many() { + ucmd() + .args(&["-P3"]) + .pipe_in("abc\ndef g\\hi 'i j \"k'") + .succeeds() + .stdout_only("abc def ghi i j \"k\n"); +} + #[test] fn xargs_null() { ucmd() @@ -345,6 +363,15 @@ fn xargs_exec_with_signal() { ); } +#[test] +fn xargs_exec_negative_parallel() { + ucmd() + .args(&["-P=-1"]) + .fails_with_code(1) + .stderr_contains("Error:") + .no_stdout(); +} + #[test] fn xargs_exec_not_found() { ucmd()