diff --git a/.github/workflows/wasi.yml b/.github/workflows/wasi.yml index 2b6555ed00c..f677481df17 100644 --- a/.github/workflows/wasi.yml +++ b/.github/workflows/wasi.yml @@ -45,6 +45,11 @@ jobs: echo "CC_wasm32_${abi}=$HOME/${WASI_SDK}/bin/clang" >> $GITHUB_ENV echo "AR_wasm32_${abi}=$HOME/${WASI_SDK}/bin/llvm-ar" >> $GITHUB_ENV done + - name: Check sort for threaded WASI + if: matrix.job.target == 'wasm32-wasip1' + run: | + rustup target add wasm32-wasip1-threads + cargo check --target wasm32-wasip1-threads --no-default-features -p uu_sort - name: Install wasmtime run: | curl https://wasmtime.dev/install.sh -sSf | bash @@ -75,7 +80,8 @@ jobs: # arch b2sum cksum csplit date dir dircolors fmt join # ls md5sum mkdir mv nproc pathchk pr printenv ptx pwd readlink # realpath rm rmdir seq sha1sum sha224sum sha256sum sha384sum - # sha512sum shred sleep sort split tsort uname uniq vdir + # sha512sum shred sleep split tsort uname uniq + # vdir UUTESTS_BINARY_PATH="$(pwd)/target/${{ matrix.job.target }}/debug/coreutils.wasm" \ UUTESTS_WASM_RUNNER=wasmtime \ cargo test --test tests -- \ @@ -84,7 +90,7 @@ jobs: test_cat:: test_comm:: test_cut:: test_dirname:: test_echo:: \ test_expand:: test_expr:: test_factor:: test_false:: test_fold:: \ test_head:: test_link:: test_ln:: test_nl:: test_numfmt:: \ - test_od:: test_paste:: test_printf:: test_shuf:: test_sum:: \ - test_tail:: test_tee:: test_test:: test_touch:: test_tr:: \ + test_od:: test_paste:: test_printf:: test_shuf:: test_sort:: \ + test_sum:: test_tail:: test_tee:: test_test:: test_touch:: test_tr:: \ test_true:: test_truncate:: \ test_unexpand:: test_unlink:: test_wc:: test_yes:: diff --git a/src/uu/sort/Cargo.toml b/src/uu/sort/Cargo.toml index eaa4251b576..1848892160a 100644 --- a/src/uu/sort/Cargo.toml +++ b/src/uu/sort/Cargo.toml @@ -28,7 +28,6 @@ clap = { workspace = true } itertools = { workspace = true } memchr = { workspace = true } rand = { workspace = true } -rayon = { workspace = true } self_cell = { workspace = true } tempfile = { workspace = true } thiserror = { workspace = true } @@ -46,6 +45,12 @@ foldhash = { workspace = true } [target.'cfg(all(unix, not(any(target_os = "redox", target_os = "fuchsia", target_os = "haiku", target_os = "solaris", target_os = "illumos"))))'.dependencies] rustix = { workspace = true, features = ["system", "process"] } +[target.'cfg(not(target_os = "wasi"))'.dependencies] +rayon = { workspace = true } + +[target.wasm32-wasip1-threads.dependencies] +rayon = { workspace = true } + [target.'cfg(not(any(target_os = "redox", target_os = "wasi")))'.dependencies] ctrlc = { workspace = true } diff --git a/src/uu/sort/build.rs b/src/uu/sort/build.rs new file mode 100644 index 00000000000..ce8fa4c335c --- /dev/null +++ b/src/uu/sort/build.rs @@ -0,0 +1,16 @@ +fn main() { + // Set a short alias for the WASI-without-threads configuration so that + // source files can use `#[cfg(wasi_no_threads)]`. + println!("cargo::rustc-check-cfg=cfg(wasi_no_threads)"); + + let target_os = std::env::var("CARGO_CFG_TARGET_OS").unwrap_or_default(); + let target = std::env::var("TARGET").unwrap_or_default(); + + // Rust currently exposes the same cfg set for the threaded and + // single-threaded WASIp1 targets, so the known threaded target must be + // selected explicitly. This also matches the target-specific Rayon + // dependency in Cargo.toml. + if target_os == "wasi" && target != "wasm32-wasip1-threads" { + println!("cargo::rustc-cfg=wasi_no_threads"); + } +} diff --git a/src/uu/sort/src/check/mod.rs b/src/uu/sort/src/check/mod.rs new file mode 100644 index 00000000000..0a6e9255419 --- /dev/null +++ b/src/uu/sort/src/check/mod.rs @@ -0,0 +1,44 @@ +// This file is part of the uutils coreutils package. +// +// For the full copyright and license information, please view the LICENSE +// file that was distributed with this source code. + +//! Check if a file is ordered. +//! +//! On most platforms this uses a multi-threaded reader. On WASI without +//! threads, a synchronous variant is used instead. The two implementations +//! live in sibling modules and are selected via cfg at the module boundary. + +use std::cmp::Ordering; +use std::ffi::OsStr; + +use uucore::error::UResult; + +use crate::{GlobalSettings, open}; + +#[cfg(not(wasi_no_threads))] +mod threaded; +#[cfg(not(wasi_no_threads))] +use threaded as runner; + +#[cfg(wasi_no_threads)] +mod sync; +#[cfg(wasi_no_threads)] +use sync as runner; + +/// Check if the file at `path` is ordered. +pub fn check(path: &OsStr, settings: &GlobalSettings) -> UResult<()> { + let max_allowed_cmp = if settings.unique { + Ordering::Less + } else { + Ordering::Equal + }; + let file = open(path)?; + let chunk_size = if settings.buffer_size < 100 * 1024 { + settings.buffer_size + } else { + 100 * 1024 + }; + + runner::check(path, settings, max_allowed_cmp, file, chunk_size) +} diff --git a/src/uu/sort/src/check/sync.rs b/src/uu/sort/src/check/sync.rs new file mode 100644 index 00000000000..1cb338caacb --- /dev/null +++ b/src/uu/sort/src/check/sync.rs @@ -0,0 +1,98 @@ +// This file is part of the uutils coreutils package. +// +// For the full copyright and license information, please view the LICENSE +// file that was distributed with this source code. + +//! Synchronous ordered-file check for targets without thread support. + +use std::cmp::Ordering; +use std::ffi::OsStr; +use std::io::Read; +use std::iter; + +use itertools::Itertools; +use uucore::error::UResult; + +use crate::{ + GlobalSettings, SortError, + chunks::{self, Chunk, RecycledChunk}, + compare_by, +}; + +pub(super) fn check( + path: &OsStr, + settings: &GlobalSettings, + max_allowed_cmp: Ordering, + mut file: Box, + chunk_size: usize, +) -> UResult<()> { + let separator = settings.line_ending.into(); + let mut carry_over = vec![]; + let mut prev_chunk: Option = None; + let mut spare_recycled: Option = None; + let mut line_idx = 0; + + loop { + let recycled = spare_recycled + .take() + .unwrap_or_else(|| RecycledChunk::new(chunk_size)); + + let (chunk, should_continue) = chunks::read_to_chunk( + recycled, + None, + &mut carry_over, + &mut file, + &mut iter::empty(), + separator, + settings, + )?; + + let Some(chunk) = chunk else { + break; + }; + + line_idx += 1; + if let Some(prev) = prev_chunk.take() { + let prev_last = prev.lines().last().unwrap(); + let new_first = chunk.lines().first().unwrap(); + + if compare_by( + prev_last, + new_first, + settings, + prev.line_data(), + chunk.line_data(), + ) > max_allowed_cmp + { + return Err(SortError::Disorder { + file: path.to_owned(), + line_number: line_idx, + line: String::from_utf8_lossy(new_first.line).into_owned(), + silent: settings.check_silent, + } + .into()); + } + spare_recycled = Some(prev.recycle()); + } + + for (a, b) in chunk.lines().iter().tuple_windows() { + line_idx += 1; + if compare_by(a, b, settings, chunk.line_data(), chunk.line_data()) > max_allowed_cmp { + return Err(SortError::Disorder { + file: path.to_owned(), + line_number: line_idx, + line: String::from_utf8_lossy(b.line).into_owned(), + silent: settings.check_silent, + } + .into()); + } + } + + prev_chunk = Some(chunk); + + if !should_continue { + break; + } + } + Ok(()) +} diff --git a/src/uu/sort/src/check.rs b/src/uu/sort/src/check/threaded.rs similarity index 61% rename from src/uu/sort/src/check.rs rename to src/uu/sort/src/check/threaded.rs index a826bc75507..5f2ac6e3ad8 100644 --- a/src/uu/sort/src/check.rs +++ b/src/uu/sort/src/check/threaded.rs @@ -3,38 +3,32 @@ // For the full copyright and license information, please view the LICENSE // file that was distributed with this source code. -//! Check if a file is ordered +//! Multi-threaded ordered-file check: a reader thread streams chunks while +//! the main thread compares the boundary between consecutive chunks. + +use std::cmp::Ordering; +use std::ffi::OsStr; +use std::io::Read; +use std::iter; +use std::sync::mpsc::{Receiver, SyncSender, sync_channel}; +use std::thread; + +use itertools::Itertools; +use uucore::error::UResult; use crate::{ GlobalSettings, SortError, chunks::{self, Chunk, RecycledChunk}, - compare_by, open, -}; -use itertools::Itertools; -use std::{ - cmp::Ordering, - ffi::OsStr, - io::Read, - iter, - sync::mpsc::{Receiver, SyncSender, sync_channel}, - thread, + compare_by, }; -use uucore::error::UResult; -/// Check if the file at `path` is ordered. -/// -/// # Returns -/// -/// The code we should exit with. -pub fn check(path: &OsStr, settings: &GlobalSettings) -> UResult<()> { - let max_allowed_cmp = if settings.unique { - // If `unique` is enabled, the previous line must compare _less_ to the next one. - Ordering::Less - } else { - // Otherwise, the line previous line must compare _less or equal_ to the next one. - Ordering::Equal - }; - let file = open(path)?; +pub(super) fn check( + path: &OsStr, + settings: &GlobalSettings, + max_allowed_cmp: Ordering, + file: Box, + chunk_size: usize, +) -> UResult<()> { let (recycled_sender, recycled_receiver) = sync_channel(2); let (loaded_sender, loaded_receiver) = sync_channel(2); thread::spawn({ @@ -42,28 +36,17 @@ pub fn check(path: &OsStr, settings: &GlobalSettings) -> UResult<()> { move || reader(file, &recycled_receiver, &loaded_sender, &settings) }); for _ in 0..2 { - let _ = recycled_sender.send(RecycledChunk::new(if settings.buffer_size < 100 * 1024 { - // when the buffer size is smaller than 100KiB we choose it instead of the default. - // this improves testability. - settings.buffer_size - } else { - 100 * 1024 - })); + let _ = recycled_sender.send(RecycledChunk::new(chunk_size)); } let mut prev_chunk: Option = None; let mut line_idx = 0; let mut result: UResult<()> = Ok(()); - // Note that we iterate over a reference, so that `loaded_receiver` is still alive - // once we stop: `chunks::read` unwraps its `send`, so dropping our end while the - // reader thread is still going would panic it. Since we stop at the *first* - // disorder, the reader is usually still working at that point, so we shut it down - // in an orderly fashion below instead of just dropping our end. + // Keep the receiver alive after the first disorder so the reader's in-flight + // send can complete while the channel is drained below. 'outer: for chunk in &loaded_receiver { line_idx += 1; if let Some(prev_chunk) = prev_chunk.take() { - // Check if the first element of the new chunk is greater than the last - // element from the previous chunk let prev_last = prev_chunk.lines().last().unwrap(); let new_first = chunk.lines().first().unwrap(); @@ -103,11 +86,6 @@ pub fn check(path: &OsStr, settings: &GlobalSettings) -> UResult<()> { prev_chunk = Some(chunk); } - - // Stop handing out buffers, so the reader runs out of work, then drain anything it - // has already produced. This lets its in-flight `send` complete instead of failing, - // and terminates because the reader can only own the (at most two) recycled chunks - // that are still outstanding. drop(recycled_sender); while loaded_receiver.recv().is_ok() {} diff --git a/src/uu/sort/src/chunks.rs b/src/uu/sort/src/chunks.rs index ca9efcd13e8..cf035fb23ac 100644 --- a/src/uu/sort/src/chunks.rs +++ b/src/uu/sort/src/chunks.rs @@ -9,10 +9,11 @@ #![allow(dead_code)] // Ignores non-used warning for `borrow_buffer` in `Chunk` +#[cfg(not(wasi_no_threads))] +use std::sync::mpsc::SyncSender; use std::{ io::{ErrorKind, Read}, ops::Range, - sync::mpsc::SyncSender, }; use memchr::memchr_iter; @@ -156,30 +157,13 @@ impl RecycledChunk { } } -/// Read a chunk, parse lines and send them. +/// Read a chunk from the input, parse lines, and return it directly. /// -/// No empty chunk will be sent. If we reach the end of the input, `false` is returned. -/// However, if this function returns `true`, it is not guaranteed that there is still -/// input left: If the input fits _exactly_ into a buffer, we will only notice that there's -/// nothing more to read at the next invocation. In case there is no input left, nothing will -/// be sent. -/// -/// # Arguments -/// -/// (see also `read_to_chunk` for a more detailed documentation) -/// -/// * `sender`: The sender to send the lines to the sorter. -/// * `recycled_chunk`: The recycled chunk, as returned by `Chunk::recycle`. -/// (i.e. `buffer.len()` should be equal to `buffer.capacity()`) -/// * `max_buffer_size`: How big `buffer` can be. -/// * `carry_over`: The bytes that must be carried over in between invocations. -/// * `file`: The current file. -/// * `next_files`: What `file` should be updated to next. -/// * `separator`: The line separator. -/// * `settings`: The global settings. +/// Returns `(Some(chunk), should_continue)` if data was read, or +/// `(None, false)` if the input was empty. The `should_continue` flag +/// indicates whether more data may remain. #[allow(clippy::too_many_arguments)] -pub fn read( - sender: &SyncSender, +pub fn read_to_chunk( recycled_chunk: RecycledChunk, max_buffer_size: Option, carry_over: &mut Vec, @@ -187,7 +171,7 @@ pub fn read( next_files: &mut impl Iterator>, separator: u8, settings: &GlobalSettings, -) -> UResult { +) -> UResult<(Option, bool)> { let RecycledChunk { lines, selections, @@ -217,7 +201,7 @@ pub fn read( carry_over.extend_from_slice(&buffer[read..]); if read != 0 { - let payload: UResult = Chunk::try_new(buffer, |buffer| { + let chunk: UResult = Chunk::try_new(buffer, |buffer| { let selections = unsafe { // SAFETY: It is safe to transmute to an empty vector of selections with shorter lifetime. // It was only temporarily transmuted to a Vec> to make recycling possible. @@ -253,7 +237,38 @@ pub fn read( line_count_hint, }) }); - sender.send(payload?).unwrap(); + Ok((Some(chunk?), should_continue)) + } else { + Ok((None, should_continue)) + } +} + +/// Read a chunk, parse lines and send them via channel. +/// +/// Wrapper around [`read_to_chunk`] for the threaded code path. +#[cfg(not(wasi_no_threads))] +#[allow(clippy::too_many_arguments)] +pub fn read( + sender: &SyncSender, + recycled_chunk: RecycledChunk, + max_buffer_size: Option, + carry_over: &mut Vec, + file: &mut T, + next_files: &mut impl Iterator>, + separator: u8, + settings: &GlobalSettings, +) -> UResult { + let (chunk, should_continue) = read_to_chunk( + recycled_chunk, + max_buffer_size, + carry_over, + file, + next_files, + separator, + settings, + )?; + if let Some(chunk) = chunk { + sender.send(chunk).unwrap(); } Ok(should_continue) } @@ -432,32 +447,3 @@ fn read_to_buffer( } } } - -/// Parse a buffer into a `ChunkContents` suitable for `Chunk::try_new`. -/// Used by the WASI single-threaded sort path. -#[cfg(target_os = "wasi")] -pub fn parse_into_chunk<'a>( - buffer: &'a [u8], - separator: u8, - settings: &GlobalSettings, -) -> ChunkContents<'a> { - let mut lines = Vec::new(); - let mut line_data = LineData::default(); - let mut token_buffer = Vec::new(); - let mut line_count_hint = 0; - parse_lines( - buffer, - &mut lines, - &mut line_data, - &mut token_buffer, - &mut line_count_hint, - separator, - settings, - ); - ChunkContents { - lines, - line_data, - token_buffer, - line_count_hint, - } -} diff --git a/src/uu/sort/src/ext_sort/mod.rs b/src/uu/sort/src/ext_sort/mod.rs index 099a4b72e62..80c0feec08c 100644 --- a/src/uu/sort/src/ext_sort/mod.rs +++ b/src/uu/sort/src/ext_sort/mod.rs @@ -6,15 +6,91 @@ //! External sort: sort large inputs that may not fit in memory. //! //! On most platforms this uses a multi-threaded chunked approach with -//! temporary files. On WASI (no threads) we fall back to an in-memory sort. +//! temporary files. On WASI without threads, a synchronous variant is used +//! instead. The two implementations live in sibling modules and are selected +//! via cfg at the module boundary. -#[cfg(not(target_os = "wasi"))] +use std::fs::File; +use std::io::Write; +use std::path::PathBuf; + +use uucore::error::UResult; + +use crate::Line; +use crate::chunks::Chunk; +use crate::merge::WriteableTmpFile; + +#[cfg(not(wasi_no_threads))] mod threaded; -#[cfg(not(target_os = "wasi"))] +#[cfg(not(wasi_no_threads))] pub use threaded::ext_sort; -#[cfg(target_os = "wasi")] -mod wasi; -#[cfg(target_os = "wasi")] -// `self::` needed to disambiguate from the `wasi` crate -pub use self::wasi::ext_sort; +#[cfg(wasi_no_threads)] +mod sync; +#[cfg(wasi_no_threads)] +pub use sync::ext_sort; + +// Note: update `test_sort::test_start_buffer` if this size is changed +// Fixed to 8 KiB (equivalent to `std::sys::io::DEFAULT_BUF_SIZE` on most targets) +pub(super) const DEFAULT_BUF_SIZE: usize = 8 * 1024; + +/// Write the lines in `chunk` to `file`, separated by `separator`. +/// `compress_prog` is used to optionally compress file contents. +pub(super) fn write( + chunk: &Chunk, + file: (File, PathBuf), + compress_prog: Option<&str>, + separator: u8, +) -> UResult { + let mut tmp_file = I::create(file, compress_prog)?; + write_lines(chunk.lines(), tmp_file.as_write(), separator)?; + tmp_file.finished_writing() +} + +fn write_lines(lines: &[Line], writer: &mut T, separator: u8) -> std::io::Result<()> { + for s in lines { + writer.write_all(s.line)?; + writer.write_all(&[separator])?; + } + Ok(()) +} + +#[cfg(test)] +mod tests { + use std::io::{self, Write}; + + use super::write_lines; + use crate::Line; + + struct FailAfterFirstWrite { + writes: usize, + } + + impl Write for FailAfterFirstWrite { + fn write(&mut self, buf: &[u8]) -> io::Result { + if self.writes == 0 { + self.writes += 1; + Ok(buf.len()) + } else { + Err(io::Error::other("write failed")) + } + } + + fn flush(&mut self) -> io::Result<()> { + Ok(()) + } + } + + #[test] + fn write_lines_propagates_write_errors() { + let lines = [Line { + line: b"line", + index: 0, + }]; + let mut writer = FailAfterFirstWrite { writes: 0 }; + + let error = write_lines(&lines, &mut writer, b'\n').unwrap_err(); + + assert_eq!(error.kind(), io::ErrorKind::Other); + } +} diff --git a/src/uu/sort/src/ext_sort/sync.rs b/src/uu/sort/src/ext_sort/sync.rs new file mode 100644 index 00000000000..6487fb7f721 --- /dev/null +++ b/src/uu/sort/src/ext_sort/sync.rs @@ -0,0 +1,179 @@ +// This file is part of the uutils coreutils package. +// +// For the full copyright and license information, please view the LICENSE +// file that was distributed with this source code. + +//! Synchronous external sort for targets without thread support +//! (e.g. `wasm32-wasip1`). +//! +//! Uses the same chunked sort-write-merge strategy as the threaded version, +//! but reads and sorts each chunk sequentially on the calling thread. + +use std::cmp::Ordering; +use std::io::{Read, Write, stderr}; + +use itertools::Itertools; +use uucore::error::UResult; + +use crate::Output; +use crate::chunks::{self, Chunk, RecycledChunk}; +use crate::merge::{self, WriteablePlainTmpFile, WriteableTmpFile}; +use crate::tmp_dir::TmpDirWrapper; +use crate::{GlobalSettings, compare_by, print_sorted, sort_by}; + +use super::{DEFAULT_BUF_SIZE, write}; + +pub fn ext_sort( + files: &mut impl Iterator>>, + settings: &GlobalSettings, + output: Output, + tmp_dir: &mut TmpDirWrapper, +) -> UResult<()> { + let separator = settings.line_ending.into(); + let mut buffer_size = match settings.buffer_size { + size if size <= 512 * 1024 * 1024 => size, + size => size / 2, + }; + if !settings.buffer_size_is_explicit { + buffer_size = buffer_size.max(8 * 1024 * 1024); + } + + if settings.compress_prog.is_some() { + let _ = writeln!( + stderr(), + "sort: warning: --compress-program is ignored on this platform" + ); + } + + let mut file = files.next().unwrap()?; + let mut carry_over = vec![]; + + // Read and sort first chunk. + let (first, cont) = chunks::read_to_chunk( + RecycledChunk::new(buffer_size.min(DEFAULT_BUF_SIZE)), + Some(buffer_size), + &mut carry_over, + &mut file, + files, + separator, + settings, + )?; + let Some(mut first) = first else { + return Ok(()); // empty input + }; + first.with_dependent_mut(|_, c| sort_by(&mut c.lines, settings, &c.line_data)); + + if !cont { + // All input fits in one chunk. + return print_chunk(&first, settings, output); + } + + // Read and sort second chunk. + let (second, cont) = chunks::read_to_chunk( + RecycledChunk::new(buffer_size.min(DEFAULT_BUF_SIZE)), + Some(buffer_size), + &mut carry_over, + &mut file, + files, + separator, + settings, + )?; + let Some(mut second) = second else { + return print_chunk(&first, settings, output); + }; + second.with_dependent_mut(|_, c| sort_by(&mut c.lines, settings, &c.line_data)); + + if !cont { + // All input fits in two chunks — merge in memory. + return print_two_chunks(first, second, settings, output); + } + + // More than two chunks: write sorted chunks to temp files, then merge. + let mut tmp_files: Vec<::Closed> = vec![]; + + tmp_files.push(write::( + &first, + tmp_dir.next_file()?, + settings.compress_prog.as_deref(), + separator, + )?); + drop(first); + + tmp_files.push(write::( + &second, + tmp_dir.next_file()?, + settings.compress_prog.as_deref(), + separator, + )?); + let mut recycled = second.recycle(); + + loop { + let (chunk, cont) = chunks::read_to_chunk( + recycled, + None, + &mut carry_over, + &mut file, + files, + separator, + settings, + )?; + let Some(mut chunk) = chunk else { break }; + chunk.with_dependent_mut(|_, c| sort_by(&mut c.lines, settings, &c.line_data)); + tmp_files.push(write::( + &chunk, + tmp_dir.next_file()?, + settings.compress_prog.as_deref(), + separator, + )?); + recycled = chunk.recycle(); + if !cont { + break; + } + } + + merge::merge_with_file_limit::<_, _, WriteablePlainTmpFile>( + tmp_files.into_iter().map(merge::ClosedTmpFile::reopen), + settings, + output, + tmp_dir, + ) +} + +/// Print a single sorted chunk. +fn print_chunk(chunk: &Chunk, settings: &GlobalSettings, output: Output) -> UResult<()> { + if settings.unique { + print_sorted( + chunk.lines().iter().dedup_by(|a, b| { + compare_by(a, b, settings, chunk.line_data(), chunk.line_data()) == Ordering::Equal + }), + settings, + output, + ) + } else { + print_sorted(chunk.lines().iter(), settings, output) + } +} + +/// Merge two in-memory chunks and print. +fn print_two_chunks(a: Chunk, b: Chunk, settings: &GlobalSettings, output: Output) -> UResult<()> { + let merged_iter = a.lines().iter().map(|line| (line, &a)).merge_by( + b.lines().iter().map(|line| (line, &b)), + |(line_a, a), (line_b, b)| { + compare_by(line_a, line_b, settings, a.line_data(), b.line_data()) != Ordering::Greater + }, + ); + if settings.unique { + print_sorted( + merged_iter + .dedup_by(|(line_a, a), (line_b, b)| { + compare_by(line_a, line_b, settings, a.line_data(), b.line_data()) + == Ordering::Equal + }) + .map(|(line, _)| line), + settings, + output, + ) + } else { + print_sorted(merged_iter.map(|(line, _)| line), settings, output) + } +} diff --git a/src/uu/sort/src/ext_sort/threaded.rs b/src/uu/sort/src/ext_sort/threaded.rs index 7dd089d0fe8..d0d0d02ddb8 100644 --- a/src/uu/sort/src/ext_sort/threaded.rs +++ b/src/uu/sort/src/ext_sort/threaded.rs @@ -7,9 +7,7 @@ //! thread, and spill to temporary files when memory is exceeded. use std::cmp::Ordering; -use std::fs::File; use std::io::{Read, Write, stderr}; -use std::path::PathBuf; use std::sync::mpsc::{Receiver, SyncSender}; use std::thread; @@ -17,20 +15,12 @@ use itertools::Itertools; use uucore::error::{UResult, strip_errno}; use crate::Output; -use crate::chunks::RecycledChunk; -use crate::merge::WriteableCompressedTmpFile; -use crate::merge::WriteablePlainTmpFile; -use crate::merge::WriteableTmpFile; +use crate::chunks::{self, Chunk, RecycledChunk}; +use crate::merge::{self, WriteableCompressedTmpFile, WriteablePlainTmpFile, WriteableTmpFile}; use crate::tmp_dir::TmpDirWrapper; -use crate::{ - GlobalSettings, Line, - chunks::{self, Chunk}, - compare_by, merge, print_sorted, sort_by, -}; +use crate::{GlobalSettings, compare_by, print_sorted, sort_by}; -// Note: update `test_sort::test_start_buffer` if this size is changed -// Fixed to 8 KiB (equivalent to `std::sys::io::DEFAULT_BUF_SIZE` on most targets) -const DEFAULT_BUF_SIZE: usize = 8 * 1024; +use super::{DEFAULT_BUF_SIZE, write}; /// Sort files by using auxiliary files for storing intermediate chunks (if needed), and output the result. /// @@ -283,23 +273,3 @@ fn read_write_loop( } } } - -/// Write the lines in `chunk` to `file`, separated by `separator`. -/// `compress_prog` is used to optionally compress file contents. -fn write( - chunk: &Chunk, - file: (File, PathBuf), - compress_prog: Option<&str>, - separator: u8, -) -> UResult { - let mut tmp_file = I::create(file, compress_prog)?; - write_lines(chunk.lines(), tmp_file.as_write(), separator); - tmp_file.finished_writing() -} - -fn write_lines(lines: &[Line], writer: &mut T, separator: u8) { - for s in lines { - writer.write_all(s.line).unwrap(); - writer.write_all(&[separator]).unwrap(); - } -} diff --git a/src/uu/sort/src/ext_sort/wasi.rs b/src/uu/sort/src/ext_sort/wasi.rs deleted file mode 100644 index 50bd5f63033..00000000000 --- a/src/uu/sort/src/ext_sort/wasi.rs +++ /dev/null @@ -1,59 +0,0 @@ -// This file is part of the uutils coreutils package. -// -// For the full copyright and license information, please view the LICENSE -// file that was distributed with this source code. - -//! WASI single-threaded sort: read all input into memory, sort, and output. -//! Threads are not available on WASI, so we bypass the chunked/threaded path. - -use std::cmp::Ordering; -use std::io::Read; - -use itertools::Itertools; -use uucore::error::UResult; - -use crate::Output; -use crate::chunks::{self, Chunk}; -use crate::tmp_dir::TmpDirWrapper; -use crate::{GlobalSettings, compare_by, print_sorted, sort_by}; - -/// Sort files by reading all input into memory, sorting in a single thread, and outputting directly. -pub fn ext_sort( - files: &mut impl Iterator>>, - settings: &GlobalSettings, - output: Output, - _tmp_dir: &mut TmpDirWrapper, -) -> UResult<()> { - let separator = settings.line_ending.into(); - // Read all input into memory at once. Unlike the threaded path which uses - // chunked buffered reads, WASI has no threads so we accept the memory cost. - // Note: there is no size limit here — WASI targets are expected to handle - // moderately sized inputs; very large files may cause OOM. - let mut input = Vec::new(); - for file in files { - file?.read_to_end(&mut input)?; - } - if input.is_empty() { - return Ok(()); - } - let mut chunk = Chunk::try_new(input, |buffer| { - Ok::<_, Box>(chunks::parse_into_chunk( - buffer, separator, settings, - )) - })?; - chunk.with_dependent_mut(|_, contents| { - sort_by(&mut contents.lines, settings, &contents.line_data); - }); - if settings.unique { - print_sorted( - chunk.lines().iter().dedup_by(|a, b| { - compare_by(a, b, settings, chunk.line_data(), chunk.line_data()) == Ordering::Equal - }), - settings, - output, - )?; - } else { - print_sorted(chunk.lines().iter(), settings, output)?; - } - Ok(()) -} diff --git a/src/uu/sort/src/merge.rs b/src/uu/sort/src/merge/mod.rs similarity index 50% rename from src/uu/sort/src/merge.rs rename to src/uu/sort/src/merge/mod.rs index a1962af4aa6..47c14e10828 100644 --- a/src/uu/sort/src/merge.rs +++ b/src/uu/sort/src/merge/mod.rs @@ -4,47 +4,65 @@ // file that was distributed with this source code. //! Merge already sorted files. //! -//! We achieve performance by splitting the tasks of sorting and writing, and reading and parsing between two threads. -//! The threads communicate over channels. There's one channel per file in the direction reader -> sorter, but only -//! one channel from the sorter back to the reader. The channels to the sorter are used to send the read chunks. -//! The sorter reads the next chunk from the channel whenever it needs the next chunk after running out of lines -//! from the previous read of the file. The channel back from the sorter to the reader has two purposes: To allow the reader -//! to reuse memory allocations and to tell the reader which file to read from next. +//! On most platforms this uses a multi-threaded reader/merger setup. On WASI +//! without threads, a synchronous variant is used instead. The two +//! implementations live in sibling modules and are selected via cfg at the +//! module boundary. use std::{ - cmp::Ordering, - collections::BinaryHeap, ffi::{OsStr, OsString}, fs::{self, File}, - io::{BufWriter, Read, Write}, - iter, + io::{self, BufWriter, Read, Write}, path::{Path, PathBuf}, process::{Child, ChildStdin, ChildStdout, Command, Stdio}, rc::Rc, - sync::mpsc::{Receiver, Sender, SyncSender, TryRecvError, channel, sync_channel}, - thread::{self, JoinHandle}, }; +use uucore::display::Quotable; use uucore::error::{FromIo, UResult}; +use uucore::translate; use crate::{ - GlobalSettings, Output, SortError, - chunks::{self, Chunk, RecycledChunk}, - compare_by, current_open_fd_count, fd_soft_limit, open, - tmp_dir::TmpDirWrapper, + GlobalSettings, Output, STDIN_FILE, SortError, chunks::Chunk, current_open_fd_count, + fd_soft_limit, open, tmp_dir::TmpDirWrapper, }; +#[cfg(not(wasi_no_threads))] +mod threaded; +#[cfg(not(wasi_no_threads))] +use threaded as runner; + +#[cfg(wasi_no_threads)] +mod sync; +#[cfg(wasi_no_threads)] +use sync as runner; + /// If the output file occurs in the input files as well, copy the contents of the output file -/// and replace its occurrences in the inputs with that copy. -fn replace_output_file_in_input_files( +/// and replace its occurrences in the inputs with that copy. Merge mode also snapshots the first +/// standard-input operand because it may be redirected from the output file. +pub(super) fn replace_output_file_in_input_files( files: &mut [OsString], output: Option<&OsStr>, + snapshot_stdin: bool, tmp_dir: &mut TmpDirWrapper, ) -> UResult<()> { let mut copy: Option = None; + let mut stdin_copied = false; if let Some(Ok(output_path)) = output.map(|path| Path::new(path).canonicalize()) { for file in files { - if Path::new(file) + if file == STDIN_FILE { + if snapshot_stdin && !stdin_copied { + let (mut copy_file, copy_path) = tmp_dir.next_file()?; + io::copy(&mut io::stdin().lock(), &mut copy_file).map_err(|error| { + SortError::ReadFailed { + path: PathBuf::from(STDIN_FILE), + error, + } + })?; + *file = copy_path.into_os_string(); + stdin_copied = true; + } + } else if Path::new(file) .canonicalize() .is_ok_and(|file_path| file_path == output_path) { @@ -100,10 +118,20 @@ pub fn merge( output: Output, tmp_dir: &mut TmpDirWrapper, ) -> UResult<()> { - replace_output_file_in_input_files(files, output.as_output_name(), tmp_dir)?; let files = files .iter() .map(|file| open(file).map(|file| PlainMergeInput { inner: file })); + + if !runner::SUPPORTS_COMPRESSION && settings.compress_prog.is_some() { + let _ = writeln!( + io::stderr(), + "sort: warning: --compress-program is ignored on this platform" + ); + return merge_with_file_limit::<_, _, WriteablePlainTmpFile>( + files, settings, output, tmp_dir, + ); + } + if settings.compress_prog.is_none() { merge_with_file_limit::<_, _, WriteablePlainTmpFile>(files, settings, output, tmp_dir) } else { @@ -111,6 +139,31 @@ pub fn merge( } } +/// Merge and write to output, dispatching to the active runner. +fn do_merge_to_output( + files: impl Iterator>, + settings: &GlobalSettings, + output: Output, +) -> UResult<()> { + let output_name = output + .as_output_name() + .unwrap_or(OsStr::new("standard output")) + .to_owned(); + let ctx = || translate!("sort-error-write-failed", "output" => output_name.maybe_quote()); + let mut out = output.into_write(); + runner::merge_without_limit(files, settings)?.write_all_to(settings, &mut out)?; + flush_writer(&mut out).map_err_context(ctx) +} + +/// Merge and write to a writer, dispatching to the active runner. +fn do_merge_to_writer( + files: impl Iterator>, + settings: &GlobalSettings, + out: &mut impl Write, +) -> UResult<()> { + runner::merge_without_limit(files, settings)?.write_all_to(settings, out) +} + // Merge already sorted `MergeInput`s. pub fn merge_with_file_limit< M: MergeInput + 'static, @@ -126,8 +179,7 @@ pub fn merge_with_file_limit< debug_assert!(batch_size >= 2); if files.len() <= batch_size { - let merger = merge_without_limit(files, settings); - merger?.write_all(settings, output) + do_merge_to_output(files, settings, output) } else { let mut temporary_files = vec![]; let mut batch = Vec::with_capacity(batch_size); @@ -135,23 +187,21 @@ pub fn merge_with_file_limit< batch.push(file); if batch.len() >= batch_size { assert_eq!(batch.len(), batch_size); - let merger = merge_without_limit(batch.into_iter(), settings)?; - batch = Vec::with_capacity(batch_size); + let full_batch = std::mem::replace(&mut batch, Vec::with_capacity(batch_size)); let mut tmp_file = Tmp::create(tmp_dir.next_file()?, settings.compress_prog.as_deref())?; - merger.write_all_to(settings, tmp_file.as_write())?; + do_merge_to_writer(full_batch.into_iter(), settings, tmp_file.as_write())?; temporary_files.push(tmp_file.finished_writing()?); } } // Merge any remaining files that didn't get merged in a full batch above. if !batch.is_empty() { assert!(batch.len() < batch_size); - let merger = merge_without_limit(batch.into_iter(), settings)?; let mut tmp_file = Tmp::create(tmp_dir.next_file()?, settings.compress_prog.as_deref())?; - merger.write_all_to(settings, tmp_file.as_write())?; + do_merge_to_writer(batch.into_iter(), settings, tmp_file.as_write())?; temporary_files.push(tmp_file.finished_writing()?); } merge_with_file_limit::<_, _, Tmp>( @@ -168,279 +218,15 @@ pub fn merge_with_file_limit< } } -/// Merge files without limiting how many files are concurrently open. -/// -/// It is the responsibility of the caller to ensure that `files` yields only -/// as many files as we are allowed to open concurrently. -fn merge_without_limit>>( - files: F, - settings: &GlobalSettings, -) -> UResult> { - let (request_sender, request_receiver) = channel(); - let mut reader_files = Vec::with_capacity(files.size_hint().0); - let mut loaded_receivers = Vec::with_capacity(files.size_hint().0); - for (file_number, file) in files.enumerate() { - let (sender, receiver) = sync_channel(2); - loaded_receivers.push(receiver); - reader_files.push(Some(ReaderFile { - file: file?, - sender, - carry_over: vec![], - })); - // Send the initial chunk to trigger a read for each file - request_sender - .send((file_number, RecycledChunk::new(8 * 1024))) - .unwrap(); - } - - // Send the second chunk for each file - for file_number in 0..reader_files.len() { - request_sender - .send((file_number, RecycledChunk::new(8 * 1024))) - .unwrap(); - } - - let reader_join_handle = thread::spawn({ - let settings = settings.clone(); - move || { - reader( - &request_receiver, - &mut reader_files, - &settings, - settings.line_ending.into(), - ) - } - }); - - let mut mergeable_files = vec![]; - - for (file_number, receiver) in loaded_receivers.into_iter().enumerate() { - if let Ok(chunk) = receiver.recv() { - mergeable_files.push(MergeableFile { - current_chunk: Rc::new(chunk), - file_number, - line_idx: 0, - receiver, - settings, - }); - } - } - - Ok(FileMerger { - heap: BinaryHeap::from(mergeable_files), - request_sender, - prev: None, - reader_join_handle, - }) -} -/// The struct on the reader thread representing an input file -struct ReaderFile { - file: M, - sender: SyncSender, - carry_over: Vec, -} - -/// The function running on the reader thread. -fn reader( - recycled_receiver: &Receiver<(usize, RecycledChunk)>, - files: &mut [Option>], - settings: &GlobalSettings, - separator: u8, -) -> UResult<()> { - for (file_idx, recycled_chunk) in recycled_receiver { - if let Some(ReaderFile { - file, - sender, - carry_over, - }) = &mut files[file_idx] - { - let should_continue = chunks::read( - sender, - recycled_chunk, - None, - carry_over, - file.as_read(), - &mut iter::empty(), - separator, - settings, - )?; - if !should_continue { - // Remove the file from the list by replacing it with `None`. - let ReaderFile { file, .. } = files[file_idx].take().unwrap(); - // Depending on the kind of the `MergeInput`, this may delete the file: - file.finished_reading()?; - } - } - } - Ok(()) -} -/// The struct on the main thread representing an input file -pub struct MergeableFile<'a> { - current_chunk: Rc, - line_idx: usize, - receiver: Receiver, - file_number: usize, - settings: &'a GlobalSettings, -} - -impl PartialEq for MergeableFile<'_> { - fn eq(&self, other: &Self) -> bool { - self.cmp(other) == Ordering::Equal - } -} - -impl Eq for MergeableFile<'_> {} - -impl PartialOrd for MergeableFile<'_> { - fn partial_cmp(&self, other: &Self) -> Option { - Some(self.cmp(other)) - } -} - -impl Ord for MergeableFile<'_> { - fn cmp(&self, other: &Self) -> Ordering { - let mut cmp = compare_by( - &self.current_chunk.lines()[self.line_idx], - &other.current_chunk.lines()[other.line_idx], - self.settings, - self.current_chunk.line_data(), - other.current_chunk.line_data(), - ); - if cmp == Ordering::Equal { - // To make sorting stable, we need to consider the file number as well, - // as lines from a file with a lower number are to be considered "earlier". - cmp = self.file_number.cmp(&other.file_number); - } - // BinaryHeap is a max heap. We use it as a min heap, so we need to reverse the ordering. - cmp.reverse() - } -} - /// A struct to keep track of the previous line we encountered. /// /// This is required for deduplication purposes. -struct PreviousLine { - chunk: Rc, - line_idx: usize, - file_number: usize, -} - -/// Merges files together. This is **not** an iterator because of lifetime problems. -struct FileMerger<'a> { - heap: BinaryHeap>, - request_sender: Sender<(usize, RecycledChunk)>, - prev: Option, - reader_join_handle: JoinHandle>, -} - -impl FileMerger<'_> { - /// Write the merged contents to the output file. - fn write_all(self, settings: &GlobalSettings, output: Output) -> UResult<()> { - let mut out = output.into_write(); - self.write_all_to(settings, &mut out) - } - - fn write_all_to(mut self, settings: &GlobalSettings, out: &mut impl Write) -> UResult<()> { - let write_result = loop { - match self - .write_next(out, settings) - .map_err_context(|| "write failed".into()) - { - Ok(true) => (), - Ok(false) => break Ok(()), - Err(error) => { - // Don't return yet: we still have to shut the reader thread down in an - // orderly fashion below. Returning here would drop our receivers while - // the reader is still sending, and `chunks::read` unwraps that send. - break Err(error); - } - } - }; - - let Self { - heap, - request_sender, - reader_join_handle, - .. - } = self; - - // Stop asking for chunks, so the reader runs out of work and returns. - drop(request_sender); - // Until it does, keep draining the files it might still be sending to. We have to - // poll all of them in turn rather than draining one at a time: the reader can be - // blocked on any one channel, and blocking on a different one would deadlock. - let mut files = heap.into_vec(); - while !files.is_empty() { - files.retain(|file| { - !matches!(file.receiver.try_recv(), Err(TryRecvError::Disconnected)) - }); - thread::yield_now(); - } - - let reader_result = reader_join_handle.join().unwrap(); - // A write failure is what the user needs to hear about; the reader hitting an error - // on the way down is secondary. - write_result.and(reader_result) - } - - fn write_next( - &mut self, - writer: &mut impl Write, - settings: &GlobalSettings, - ) -> std::io::Result { - if let Some(file) = self.heap.peek() { - let prev = self.prev.replace(PreviousLine { - chunk: file.current_chunk.clone(), - line_idx: file.line_idx, - file_number: file.file_number, - }); - - file.current_chunk.with_dependent(|_, contents| { - let current_line = &contents.lines[file.line_idx]; - if settings.unique - && let Some(prev) = &prev - { - let cmp = compare_by( - &prev.chunk.lines()[prev.line_idx], - current_line, - settings, - prev.chunk.line_data(), - file.current_chunk.line_data(), - ); - if cmp == Ordering::Equal { - return Ok(()); - } - } - current_line.write(writer, settings) - })?; - - let was_last_line_for_file = file.current_chunk.lines().len() == file.line_idx + 1; - - if was_last_line_for_file { - if let Ok(next_chunk) = file.receiver.recv() { - let mut file = self.heap.peek_mut().unwrap(); - file.current_chunk = Rc::new(next_chunk); - file.line_idx = 0; - } else { - self.heap.pop(); - } - } else { - // This will cause the comparison to use a different line and the heap to readjust. - self.heap.peek_mut().unwrap().line_idx += 1; - } - - if let Some(prev) = prev - && let Ok(prev_chunk) = Rc::try_unwrap(prev.chunk) - { - // If nothing is referencing the previous chunk anymore, this means that the previous line - // was the last line of the chunk. We can recycle the chunk. - self.request_sender - .send((prev.file_number, prev_chunk.recycle())) - .ok(); - } - } - Ok(!self.heap.is_empty()) - } +pub(super) struct PreviousLine { + pub chunk: Rc, + pub line_idx: usize, + // Only the threaded merger reads this back to recycle chunks. + #[cfg_attr(wasi_no_threads, allow(dead_code))] + pub file_number: usize, } /// Wait for the child to exit and check its exit code. @@ -490,6 +276,11 @@ pub struct PlainTmpMergeInput { path: PathBuf, file: File, } + +fn flush_writer(writer: &mut impl Write) -> io::Result<()> { + writer.flush() +} + impl WriteableTmpFile for WriteablePlainTmpFile { type Closed = ClosedPlainTmpFile; type InnerWrite = BufWriter; @@ -501,7 +292,8 @@ impl WriteableTmpFile for WriteablePlainTmpFile { }) } - fn finished_writing(self) -> UResult { + fn finished_writing(mut self) -> UResult { + flush_writer(&mut self.file)?; Ok(ClosedPlainTmpFile { path: self.path }) } @@ -573,7 +365,8 @@ impl WriteableTmpFile for WriteableCompressedTmpFile { }) } - fn finished_writing(self) -> UResult { + fn finished_writing(mut self) -> UResult { + flush_writer(&mut self.child_stdin)?; drop(self.child_stdin); check_child_success(self.child, &self.compress_prog)?; Ok(ClosedCompressedTmpFile { @@ -586,6 +379,7 @@ impl WriteableTmpFile for WriteableCompressedTmpFile { &mut self.child_stdin } } + impl ClosedTmpFile for ClosedCompressedTmpFile { type Reopened = CompressedTmpMergeInput; @@ -639,3 +433,33 @@ impl MergeInput for PlainMergeInput { &mut self.inner } } + +#[cfg(test)] +mod tests { + use std::io::{self, BufWriter, Write}; + + use super::flush_writer; + + struct FailOnFlush; + + impl Write for FailOnFlush { + fn write(&mut self, buf: &[u8]) -> io::Result { + Ok(buf.len()) + } + + fn flush(&mut self) -> io::Result<()> { + Err(io::Error::other("flush failed")) + } + } + + #[test] + fn flush_writer_propagates_flush_errors() { + let mut writer = BufWriter::new(FailOnFlush); + writer.write_all(b"buffered data").unwrap(); + + let error = flush_writer(&mut writer).unwrap_err(); + + assert_eq!(error.kind(), io::ErrorKind::Other); + let _ = writer.into_parts(); + } +} diff --git a/src/uu/sort/src/merge/sync.rs b/src/uu/sort/src/merge/sync.rs new file mode 100644 index 00000000000..c84d7a35a67 --- /dev/null +++ b/src/uu/sort/src/merge/sync.rs @@ -0,0 +1,225 @@ +// This file is part of the uutils coreutils package. +// +// For the full copyright and license information, please view the LICENSE +// file that was distributed with this source code. + +//! Synchronous merge for targets without thread support (e.g. wasm32-wasip1). +//! +//! Reads chunks on demand from each input on the calling thread instead of +//! using a dedicated reader thread. + +use std::{cmp::Ordering, collections::BinaryHeap, io::Write, iter, rc::Rc}; + +use uucore::error::UResult; + +use crate::{ + GlobalSettings, + chunks::{self, Chunk, RecycledChunk}, + compare_by, +}; + +use super::{MergeInput, PreviousLine}; + +pub(super) const SUPPORTS_COMPRESSION: bool = false; + +struct SyncReaderFile { + file: M, + carry_over: Vec, +} + +struct SyncMergeableFile<'a> { + current_chunk: Rc, + line_idx: usize, + file_number: usize, + settings: &'a GlobalSettings, +} + +impl PartialEq for SyncMergeableFile<'_> { + fn eq(&self, other: &Self) -> bool { + self.cmp(other) == Ordering::Equal + } +} + +impl Eq for SyncMergeableFile<'_> {} + +impl PartialOrd for SyncMergeableFile<'_> { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl Ord for SyncMergeableFile<'_> { + fn cmp(&self, other: &Self) -> Ordering { + let mut cmp = compare_by( + &self.current_chunk.lines()[self.line_idx], + &other.current_chunk.lines()[other.line_idx], + self.settings, + self.current_chunk.line_data(), + other.current_chunk.line_data(), + ); + if cmp == Ordering::Equal { + cmp = self.file_number.cmp(&other.file_number); + } + cmp.reverse() + } +} + +pub(super) struct SyncFileMerger<'a, M: MergeInput> { + heap: BinaryHeap>, + readers: Vec>>, + prev: Option, + recycled: Option, + settings: &'a GlobalSettings, +} + +impl SyncFileMerger<'_, M> { + pub(super) fn write_all_to( + mut self, + settings: &GlobalSettings, + out: &mut impl Write, + ) -> UResult<()> { + let write_result = loop { + match self.write_next(out, settings) { + Ok(true) => {} + Ok(false) => break Ok(()), + Err(error) => break Err(error), + } + }; + + // Readers finished during the merge are replaced with `None`; clean up + // any that remain after an early write failure. + let cleanup_result = self + .readers + .into_iter() + .flatten() + .try_for_each(|reader| reader.file.finished_reading()); + write_result.and(cleanup_result) + } + + fn write_next(&mut self, writer: &mut impl Write, settings: &GlobalSettings) -> UResult { + if let Some(file) = self.heap.peek() { + let prev = self.prev.replace(PreviousLine { + chunk: file.current_chunk.clone(), + line_idx: file.line_idx, + file_number: file.file_number, + }); + + file.current_chunk.with_dependent(|_, contents| { + let current_line = &contents.lines[file.line_idx]; + if settings.unique + && let Some(prev) = &prev + { + let cmp = compare_by( + &prev.chunk.lines()[prev.line_idx], + current_line, + settings, + prev.chunk.line_data(), + file.current_chunk.line_data(), + ); + if cmp == Ordering::Equal { + return Ok(()); + } + } + current_line.write(writer, settings) + })?; + + let was_last = file.current_chunk.lines().len() == file.line_idx + 1; + let file_number = file.file_number; + + if was_last { + let separator = self.settings.line_ending.into(); + let recycled = self + .recycled + .take() + .unwrap_or_else(|| RecycledChunk::new(8 * 1024)); + let next_chunk = if let Some(reader) = self.readers[file_number].as_mut() { + let (chunk, should_continue) = chunks::read_to_chunk( + recycled, + None, + &mut reader.carry_over, + reader.file.as_read(), + &mut iter::empty(), + separator, + self.settings, + )?; + if !should_continue && let Some(reader) = self.readers[file_number].take() { + reader.file.finished_reading()?; + } + chunk + } else { + None + }; + + if let Some(next_chunk) = next_chunk { + let mut file = self.heap.peek_mut().unwrap(); + file.current_chunk = Rc::new(next_chunk); + file.line_idx = 0; + } else { + self.heap.pop(); + } + } else { + self.heap.peek_mut().unwrap().line_idx += 1; + } + + // Recycle the previous chunk if no other reference holds it. + if let Some(prev) = prev + && let Ok(chunk) = Rc::try_unwrap(prev.chunk) + { + self.recycled = Some(chunk.recycle()); + } + } + Ok(!self.heap.is_empty()) + } +} + +pub(super) fn merge_without_limit>>( + files: F, + settings: &GlobalSettings, +) -> UResult> { + let separator = settings.line_ending.into(); + let mut readers: Vec>> = Vec::new(); + let mut mergeable_files = Vec::new(); + + for (file_number, file) in files.enumerate() { + let mut reader = SyncReaderFile { + file: file?, + carry_over: vec![], + }; + let recycled = RecycledChunk::new(8 * 1024); + let (chunk, should_continue) = chunks::read_to_chunk( + recycled, + None, + &mut reader.carry_over, + reader.file.as_read(), + &mut iter::empty(), + separator, + settings, + )?; + + if let Some(chunk) = chunk { + mergeable_files.push(SyncMergeableFile { + current_chunk: Rc::new(chunk), + line_idx: 0, + file_number, + settings, + }); + if should_continue { + readers.push(Some(reader)); + } else { + reader.file.finished_reading()?; + readers.push(None); + } + } else { + reader.file.finished_reading()?; + readers.push(None); + } + } + + Ok(SyncFileMerger { + heap: BinaryHeap::from(mergeable_files), + readers, + prev: None, + recycled: None, + settings, + }) +} diff --git a/src/uu/sort/src/merge/threaded.rs b/src/uu/sort/src/merge/threaded.rs new file mode 100644 index 00000000000..131101be74c --- /dev/null +++ b/src/uu/sort/src/merge/threaded.rs @@ -0,0 +1,281 @@ +// This file is part of the uutils coreutils package. +// +// For the full copyright and license information, please view the LICENSE +// file that was distributed with this source code. + +//! Multi-threaded merge: a reader thread feeds chunks into the per-file +//! channels while the main thread merges the next-line heap. + +use std::{ + cmp::Ordering, + collections::BinaryHeap, + io::Write, + iter, + rc::Rc, + sync::mpsc::{Receiver, Sender, SyncSender, TryRecvError, channel, sync_channel}, + thread::{self, JoinHandle}, +}; + +use uucore::error::{FromIo, UResult}; + +use crate::{ + GlobalSettings, + chunks::{self, Chunk, RecycledChunk}, + compare_by, +}; + +use super::{MergeInput, PreviousLine}; + +pub(super) const SUPPORTS_COMPRESSION: bool = true; + +/// Merge files without limiting how many files are concurrently open. +/// +/// It is the responsibility of the caller to ensure that `files` yields only +/// as many files as we are allowed to open concurrently. +pub(super) fn merge_without_limit>>( + files: F, + settings: &GlobalSettings, +) -> UResult> { + let (request_sender, request_receiver) = channel(); + let mut reader_files = Vec::with_capacity(files.size_hint().0); + let mut loaded_receivers = Vec::with_capacity(files.size_hint().0); + for (file_number, file) in files.enumerate() { + let (sender, receiver) = sync_channel(2); + loaded_receivers.push(receiver); + reader_files.push(Some(ReaderFile { + file: file?, + sender, + carry_over: vec![], + })); + // Send the initial chunk to trigger a read for each file + request_sender + .send((file_number, RecycledChunk::new(8 * 1024))) + .unwrap(); + } + + // Send the second chunk for each file + for file_number in 0..reader_files.len() { + request_sender + .send((file_number, RecycledChunk::new(8 * 1024))) + .unwrap(); + } + + let reader_join_handle = thread::spawn({ + let settings = settings.clone(); + move || { + reader( + &request_receiver, + &mut reader_files, + &settings, + settings.line_ending.into(), + ) + } + }); + + let mut mergeable_files = vec![]; + + for (file_number, receiver) in loaded_receivers.into_iter().enumerate() { + if let Ok(chunk) = receiver.recv() { + mergeable_files.push(MergeableFile { + current_chunk: Rc::new(chunk), + file_number, + line_idx: 0, + receiver, + settings, + }); + } + } + + Ok(FileMerger { + heap: BinaryHeap::from(mergeable_files), + request_sender, + prev: None, + reader_join_handle, + }) +} + +/// The struct on the reader thread representing an input file +struct ReaderFile { + file: M, + sender: SyncSender, + carry_over: Vec, +} + +/// The function running on the reader thread. +fn reader( + recycled_receiver: &Receiver<(usize, RecycledChunk)>, + files: &mut [Option>], + settings: &GlobalSettings, + separator: u8, +) -> UResult<()> { + for (file_idx, recycled_chunk) in recycled_receiver { + if let Some(ReaderFile { + file, + sender, + carry_over, + }) = &mut files[file_idx] + { + let should_continue = chunks::read( + sender, + recycled_chunk, + None, + carry_over, + file.as_read(), + &mut iter::empty(), + separator, + settings, + )?; + if !should_continue { + // Remove the file from the list by replacing it with `None`. + let ReaderFile { file, .. } = files[file_idx].take().unwrap(); + // Depending on the kind of the `MergeInput`, this may delete the file: + file.finished_reading()?; + } + } + } + Ok(()) +} + +/// The struct on the main thread representing an input file +pub(super) struct MergeableFile<'a> { + current_chunk: Rc, + line_idx: usize, + receiver: Receiver, + file_number: usize, + settings: &'a GlobalSettings, +} + +impl PartialEq for MergeableFile<'_> { + fn eq(&self, other: &Self) -> bool { + self.cmp(other) == Ordering::Equal + } +} + +impl Eq for MergeableFile<'_> {} + +impl PartialOrd for MergeableFile<'_> { + fn partial_cmp(&self, other: &Self) -> Option { + Some(self.cmp(other)) + } +} + +impl Ord for MergeableFile<'_> { + fn cmp(&self, other: &Self) -> Ordering { + let mut cmp = compare_by( + &self.current_chunk.lines()[self.line_idx], + &other.current_chunk.lines()[other.line_idx], + self.settings, + self.current_chunk.line_data(), + other.current_chunk.line_data(), + ); + if cmp == Ordering::Equal { + cmp = self.file_number.cmp(&other.file_number); + } + cmp.reverse() + } +} + +/// Merges files together. This is **not** an iterator because of lifetime problems. +pub(super) struct FileMerger<'a> { + heap: BinaryHeap>, + request_sender: Sender<(usize, RecycledChunk)>, + prev: Option, + reader_join_handle: JoinHandle>, +} + +impl FileMerger<'_> { + pub(super) fn write_all_to( + mut self, + settings: &GlobalSettings, + out: &mut impl Write, + ) -> UResult<()> { + let write_result = loop { + match self + .write_next(out, settings) + .map_err_context(|| "write failed".into()) + { + Ok(true) => (), + Ok(false) => break Ok(()), + Err(error) => break Err(error), + } + }; + + let Self { + heap, + request_sender, + reader_join_handle, + .. + } = self; + + drop(request_sender); + let mut files = heap.into_vec(); + while !files.is_empty() { + files.retain(|file| { + !matches!(file.receiver.try_recv(), Err(TryRecvError::Disconnected)) + }); + thread::yield_now(); + } + + let reader_result = reader_join_handle.join().unwrap(); + write_result.and(reader_result) + } + + fn write_next( + &mut self, + writer: &mut impl Write, + settings: &GlobalSettings, + ) -> std::io::Result { + if let Some(file) = self.heap.peek() { + let prev = self.prev.replace(PreviousLine { + chunk: file.current_chunk.clone(), + line_idx: file.line_idx, + file_number: file.file_number, + }); + + file.current_chunk.with_dependent(|_, contents| { + let current_line = &contents.lines[file.line_idx]; + if settings.unique + && let Some(prev) = &prev + { + let cmp = compare_by( + &prev.chunk.lines()[prev.line_idx], + current_line, + settings, + prev.chunk.line_data(), + file.current_chunk.line_data(), + ); + if cmp == Ordering::Equal { + return Ok(()); + } + } + current_line.write(writer, settings) + })?; + + let was_last_line_for_file = file.current_chunk.lines().len() == file.line_idx + 1; + + if was_last_line_for_file { + if let Ok(next_chunk) = file.receiver.recv() { + let mut file = self.heap.peek_mut().unwrap(); + file.current_chunk = Rc::new(next_chunk); + file.line_idx = 0; + } else { + self.heap.pop(); + } + } else { + // This will cause the comparison to use a different line and the heap to readjust. + self.heap.peek_mut().unwrap().line_idx += 1; + } + + if let Some(prev) = prev + && let Ok(prev_chunk) = Rc::try_unwrap(prev.chunk) + { + // If nothing is referencing the previous chunk anymore, this means that the previous line + // was the last line of the chunk. We can recycle the chunk. + self.request_sender + .send((prev.file_number, prev_chunk.recycle())) + .ok(); + } + } + Ok(!self.heap.is_empty()) + } +} diff --git a/src/uu/sort/src/parallel.rs b/src/uu/sort/src/parallel.rs new file mode 100644 index 00000000000..c0c7aeddc6c --- /dev/null +++ b/src/uu/sort/src/parallel.rs @@ -0,0 +1,59 @@ +// This file is part of the uutils coreutils package. +// +// For the full copyright and license information, please view the LICENSE +// file that was distributed with this source code. + +//! Parallel-or-sequential sort helpers and thread-pool initialization. +//! +//! On targets without thread support (such as `wasm32-wasip1`) these +//! fall back to the sequential `[T]::sort_*` methods and a no-op pool init. +//! On every other target they use rayon's parallel sorts. + +#[cfg(not(wasi_no_threads))] +mod imp { + use std::cmp::Ordering; + use std::num::NonZero; + + use rayon::slice::ParallelSliceMut; + + pub fn sort_by(slice: &mut [T], cmp: impl Fn(&T, &T) -> Ordering + Sync) { + slice.par_sort_by(cmp); + } + + pub fn sort_unstable_by(slice: &mut [T], cmp: impl Fn(&T, &T) -> Ordering + Sync) { + slice.par_sort_unstable_by(cmp); + } + + pub fn init_thread_pool(num_threads: Option) { + let num_threads = num_threads.map_or_else( + || std::thread::available_parallelism().map_or(1, NonZero::get), + |n| n as usize, + ); + let _ = rayon::ThreadPoolBuilder::new() + .num_threads(num_threads) + .build_global(); + } +} + +#[cfg(wasi_no_threads)] +mod imp { + use std::cmp::Ordering; + + // The `Send`/`Sync` bounds mirror the parallel implementation so that call + // sites compile identically on both targets. They are stricter than the + // underlying `[T]::sort_*` methods require, but every caller in this crate + // already satisfies them. + pub fn sort_by(slice: &mut [T], cmp: impl Fn(&T, &T) -> Ordering + Sync) { + slice.sort_by(cmp); + } + + pub fn sort_unstable_by(slice: &mut [T], cmp: impl Fn(&T, &T) -> Ordering + Sync) { + slice.sort_unstable_by(cmp); + } + + pub fn init_thread_pool(_num_threads: Option) { + // No-op: there is no thread pool on this target, so --parallel is ignored. + } +} + +pub use imp::{init_thread_pool, sort_by, sort_unstable_by}; diff --git a/src/uu/sort/src/sort.rs b/src/uu/sort/src/sort.rs index f47f457eda9..cefda0398e3 100644 --- a/src/uu/sort/src/sort.rs +++ b/src/uu/sort/src/sort.rs @@ -18,6 +18,7 @@ mod diagnostics; mod ext_sort; mod merge; mod numeric_str_cmp; +mod parallel; mod tmp_dir; use bigdecimal::BigDecimal; @@ -30,8 +31,6 @@ use foldhash::fast::FoldHasher; use foldhash::{HashMap, SharedSeed}; use numeric_str_cmp::{NumInfo, NumInfoParseSettings, human_numeric_str_cmp, numeric_str_cmp}; use rand::{RngExt as _, rng}; -#[cfg(not(target_os = "wasi"))] -use rayon::slice::ParallelSliceMut; use std::cmp::Ordering; use std::env; use std::ffi::{OsStr, OsString}; @@ -2268,17 +2267,8 @@ pub fn uumain(args: impl uucore::Args) -> UResult<()> { settings.ignore_non_printing = ignore_non_printing; settings.ignore_case = ignore_case; - // WASI doesn't support threads, so we ignore the corresponding option - #[cfg(not(target_os = "wasi"))] - { - let threads = matches - .get_one::(options::PARALLEL) - .copied() - .unwrap_or_else(|| std::thread::available_parallelism().map_or(1, |n| n.get() as u64)); - let _ = rayon::ThreadPoolBuilder::new() - .num_threads(threads as usize) - .build_global(); - } + // On targets without thread support this is a no-op and --parallel is ignored. + parallel::init_thread_pool(matches.get_one::(options::PARALLEL).copied()); if let Some(size_str) = matches.get_one::(options::BUF_SIZE) { settings.buffer_size = GlobalSettings::parse_byte_count(size_str).map_err(|error| { @@ -2297,22 +2287,22 @@ pub fn uumain(args: impl uucore::Args) -> UResult<()> { settings.buffer_size_is_explicit = false; } - let mut tmp_dir = TmpDirWrapper::new(matches.get_one::(options::TMP_DIR).map_or_else( - || { - // WASI does not support std::env::temp_dir() — it panics with - // "no filesystem on wasm". Use /tmp as a nominal fallback; - // the WASI ext_sort path never actually creates temp files. - #[cfg(target_os = "wasi")] - { - PathBuf::from("/tmp") - } - #[cfg(not(target_os = "wasi"))] - { - env::temp_dir() - } - }, - PathBuf::from, - )); + let mut tmp_dir = TmpDirWrapper::new( + matches + .get_one::(options::TMP_DIR) + .map(PathBuf::from) + .or_else(|| env::var_os("TMPDIR").map(PathBuf::from)) + .unwrap_or_else(|| { + #[cfg(target_os = "wasi")] + { + uucore::fs::wasi_default_tmp_dir() + } + #[cfg(not(target_os = "wasi"))] + { + env::temp_dir() + } + }), + ); settings.compress_prog = matches .get_one::(options::COMPRESS_PROG) @@ -2470,6 +2460,17 @@ pub fn uumain(args: impl uucore::Args) -> UResult<()> { }); } + if !settings.check { + merge::replace_output_file_in_input_files( + &mut files, + matches + .get_one::(options::OUTPUT) + .map(OsString::as_os_str), + settings.merge, + &mut tmp_dir, + )?; + } + let opened_inputs = if settings.merge || settings.check { Vec::new() } else { @@ -2768,18 +2769,10 @@ fn exec( fn sort_by<'a>(unsorted: &mut Vec>, settings: &GlobalSettings, line_data: &LineData<'a>) { let cmp = |a: &Line<'a>, b: &Line<'a>| compare_by(a, b, settings, line_data, line_data); - // WASI does not support threads, so use non-parallel sort to avoid - // rayon's thread pool which triggers an unreachable trap. if settings.stable || settings.unique { - #[cfg(not(target_os = "wasi"))] - unsorted.par_sort_by(cmp); - #[cfg(target_os = "wasi")] - unsorted.sort_by(cmp); + parallel::sort_by(unsorted, cmp); } else { - #[cfg(not(target_os = "wasi"))] - unsorted.par_sort_unstable_by(cmp); - #[cfg(target_os = "wasi")] - unsorted.sort_unstable_by(cmp); + parallel::sort_unstable_by(unsorted, cmp); } } diff --git a/src/uucore/src/lib/features/fs.rs b/src/uucore/src/lib/features/fs.rs index e47ca3f7bb7..fbaa4239909 100644 --- a/src/uucore/src/lib/features/fs.rs +++ b/src/uucore/src/lib/features/fs.rs @@ -5,7 +5,7 @@ //! Set of functions to manage regular files, special files, and links. -// spell-checker:ignore backport Ioctl absolutized +// spell-checker:ignore backport Ioctl absolutized preopen #[cfg(all(unix, not(target_os = "redox")))] pub use libc::{major, makedev, minor}; @@ -238,6 +238,26 @@ pub enum ResolveMode { Logical, } +/// WASI fallback used when neither `--tmp-dir` nor `TMPDIR` is set and +/// `env::temp_dir()` would be inapplicable. +/// +/// The WASI sandbox only exposes explicitly preopened directories, and +/// `/tmp` is not one by default. This returns `/tmp` when a host preopen +/// has made it visible as a directory, and the current directory otherwise +/// — the current directory is always accessible under a preopen mapped +/// to `/`. +/// +/// Callers on WASI should prefer `--tmp-dir` and `TMPDIR` before falling +/// back to this helper. +#[cfg(target_os = "wasi")] +pub fn wasi_default_tmp_dir() -> PathBuf { + if fs::metadata("/tmp").is_ok_and(|m| m.is_dir()) { + PathBuf::from("/tmp") + } else { + PathBuf::from(".") + } +} + /// Normalize a path by removing relative information /// For example, convert 'bar/../foo/bar.txt' => 'foo/bar.txt' /// copied from `` diff --git a/tests/by-util/test_sort.rs b/tests/by-util/test_sort.rs index 44c59b90597..6140f17011c 100644 --- a/tests/by-util/test_sort.rs +++ b/tests/by-util/test_sort.rs @@ -3,7 +3,7 @@ // For the full copyright and license information, please view the LICENSE // file that was distributed with this source code. -// spell-checker:ignore (words) ints (linux) NOFILE dfgi abmon avril +// spell-checker:ignore (words) ints (linux) NOFILE dfgi abmon avril setrlimit EISDIR #![allow(clippy::cast_possible_wrap)] use std::env; @@ -37,10 +37,6 @@ fn test_helper(file_name: &str, possible_args: &[&str]) { #[test] fn test_buffer_sizes() { - #[cfg(target_os = "linux")] - let buffer_sizes = ["0", "50K", "50k", "1M", "100M", "0%", "10%"]; - // TODO Percentage sizes are not yet supported beyond Linux. - #[cfg(not(target_os = "linux"))] let buffer_sizes = ["0", "50K", "50k", "1M", "100M"]; for buffer_size in &buffer_sizes { new_ucmd!() @@ -52,7 +48,23 @@ fn test_buffer_sizes() { .stdout_is_fixture("ext_sort.expected"); } - #[cfg(not(target_pointer_width = "32"))] + // TODO Percentage sizes are not yet supported beyond Linux. A WASI guest + // also cannot inspect the Linux host's physical memory. + #[cfg(all(target_os = "linux", not(wasi_runner)))] + for buffer_size in ["0%", "10%"] { + new_ucmd!() + .arg("-n") + .arg("-S") + .arg(buffer_size) + .arg("ext_sort.txt") + .succeeds() + .stdout_is_fixture("ext_sort.expected"); + } + + // The test runner compiles for the host (often 64-bit), but the binary + // under test may be 32-bit (e.g. wasm32-wasip1), which rejects very + // large buffer sizes. + #[cfg(all(not(target_pointer_width = "32"), not(wasi_runner)))] { let buffer_sizes = ["1000G", "10T"]; for buffer_size in &buffer_sizes { @@ -92,7 +104,9 @@ fn test_invalid_buffer_size() { // A percentage can fit in a u128 while its product with the total // physical memory does not; the parser must report it as too large // rather than panicking or silently wrapping. - #[cfg(target_os = "linux")] + // The test runner is built for Linux, but the binary under test may not + // expose Linux host memory information (e.g. wasm32-wasip1). + #[cfg(all(target_os = "linux", not(wasi_runner)))] new_ucmd!() .arg("-S") .arg("340282366920938463463374607431768211455%") @@ -776,6 +790,7 @@ fn month_sort_input_expected(months: &[String]) -> (String, String) { #[test] #[cfg(unix)] +#[cfg_attr(wasi_runner, ignore = "WASI: no locale data")] fn test_month_sort_french_locale() { let locale = "fr_FR.UTF-8"; if !is_locale_available(locale) { @@ -807,6 +822,7 @@ fn test_month_sort_french_locale() { #[test] #[cfg(unix)] +#[cfg_attr(wasi_runner, ignore = "WASI: no locale data")] fn test_month_sort_hungarian_locale() { let locale = "hu_HU.UTF-8"; if !is_locale_available(locale) { @@ -838,6 +854,7 @@ fn test_month_sort_hungarian_locale() { /// E.g. "av ril" should NOT match "avril" — GNU treats it as unknown. #[test] #[cfg(unix)] +#[cfg_attr(wasi_runner, ignore = "WASI: no locale data")] fn test_month_sort_french_embedded_blanks() { let locale = "fr_FR.UTF-8"; if !is_locale_available(locale) { @@ -890,6 +907,7 @@ fn test_month_sort_french_embedded_blanks() { #[test] #[cfg(unix)] +#[cfg_attr(wasi_runner, ignore = "WASI: no locale data")] fn test_month_sort_japanese_locale() { let locale = "ja_JP.UTF-8"; if !is_locale_available(locale) { @@ -1315,6 +1333,22 @@ fn test_read_error_message() { ); } +#[test] +#[cfg(target_os = "linux")] +fn test_merge_flush_error_is_reported() { + use std::fs::File; + + let ts = TestScenario::new("sort"); + ts.fixtures.write("input.txt", "line\n"); + + let dev_full = File::create("/dev/full").expect("Failed to open /dev/full"); + ts.ucmd() + .args(&["-m", "input.txt"]) + .set_stdout(dev_full) + .fails() + .stderr_contains("No space left on device"); +} + #[test] fn test_merge_unique() { new_ucmd!() @@ -1531,6 +1565,7 @@ fn sort_empty_chunk() { #[test] #[cfg(any(target_os = "linux", target_os = "android"))] +#[cfg_attr(wasi_runner, ignore = "WASI: no subprocess spawning")] fn test_compress() { new_ucmd!() .args(&[ @@ -1547,6 +1582,7 @@ fn test_compress() { #[test] #[cfg(any(target_os = "linux", target_os = "android"))] +#[cfg_attr(wasi_runner, ignore = "WASI: no subprocess spawning")] fn test_compress_merge() { new_ucmd!() .args(&[ @@ -1570,6 +1606,7 @@ fn test_compress_merge() { #[test] #[cfg(not(target_os = "android"))] +#[cfg_attr(wasi_runner, ignore = "WASI: no subprocess spawning")] fn test_compress_fail() { let result = new_ucmd!() .args(&[ @@ -1632,7 +1669,7 @@ fn test_batch_size_too_large() { "--batch-size argument '{large_batch_size}' too large" )); - #[cfg(target_os = "linux")] + #[cfg(all(target_os = "linux", not(wasi_runner)))] new_ucmd!() .arg(format!("--batch-size={large_batch_size}")) .fails_with_code(2) @@ -1659,6 +1696,7 @@ fn test_merge_batch_size() { // TODO(#7542): Re-enable on Android once we figure out why setting limit is broken. // #[cfg(any(target_os = "linux", target_os = "android"))] #[cfg(target_os = "linux")] +#[cfg_attr(wasi_runner, ignore = "WASI: rlimit/setrlimit not supported")] fn test_merge_batch_size_with_limit() { use rlimit::Resource; // Currently need... @@ -1684,6 +1722,7 @@ fn test_merge_batch_size_with_limit() { #[test] // TODO(#7542): Re-enable on Android once we figure out why setting limit is broken. #[cfg(target_os = "linux")] +#[cfg_attr(wasi_runner, ignore = "WASI: rlimit/setrlimit not supported")] fn test_batch_size_above_fd_limit_is_rejected() { use rlimit::Resource; // Only stdin, stdout and stderr are unavailable for merge inputs, so the @@ -1702,6 +1741,7 @@ fn test_batch_size_above_fd_limit_is_rejected() { #[test] #[cfg(target_os = "linux")] +#[cfg_attr(wasi_runner, ignore = "WASI: rlimit/setrlimit not supported")] fn test_batch_size_at_fd_limit_is_accepted() { use rlimit::Resource; let limit_fd = 27; @@ -1753,6 +1793,7 @@ fn test_sigpipe_panic() { #[test] #[cfg(unix)] +#[cfg_attr(wasi_runner, ignore = "WASI: no FIFO/mkfifo support")] fn test_fifo_without_trailing_newline() { let (at, mut ucmd) = at_and_ucmd!(); at.mkfifo("FIFO"); @@ -1838,6 +1879,7 @@ fn test_verifies_files_after_keys() { #[test] #[cfg(unix)] +#[cfg_attr(wasi_runner, ignore = "WASI sandbox: host paths not visible")] fn test_verifies_input_files() { new_ucmd!() .args(&["/dev/random", "nonexistent_file"]) @@ -1898,8 +1940,53 @@ fn test_output_is_input() { assert_eq!(at.read("file"), input); } +#[test] +fn test_output_is_input_without_merge() { + let input = "a\nb\n"; + let (at, mut ucmd) = at_and_ucmd!(); + + at.write("file", input); + + ucmd.args(&["-r", "-o", "file", "file"]).succeeds(); + assert_eq!(at.read("file"), "b\na\n"); +} + +#[test] +fn test_output_is_input_via_stdin_in_merge_mode() { + let (at, mut ucmd) = at_and_ucmd!(); + at.write("file", "a\n"); + + ucmd.args(&["-m", "-o", "file", "-"]) + .set_stdin(std::fs::File::open(at.plus("file")).unwrap()) + .succeeds(); + + assert_eq!(at.read("file"), "a\n"); +} + +#[test] +fn test_merge_with_output_reads_repeated_stdin_once() { + let (at, mut ucmd) = at_and_ucmd!(); + + ucmd.args(&["-m", "-o", "file", "-", "-"]) + .pipe_in("a\n") + .succeeds(); + + assert_eq!(at.read("file"), "a\n"); +} + +#[test] +fn test_output_named_stdin_marker_reads_stdin() { + let (at, mut ucmd) = at_and_ucmd!(); + at.write("-", "old contents\n"); + + ucmd.args(&["-o", "-", "-"]).pipe_in("b\na\n").succeeds(); + + assert_eq!(at.read("-"), "a\nb\n"); +} + #[test] #[cfg(unix)] +#[cfg_attr(wasi_runner, ignore = "WASI sandbox: host paths not visible")] fn test_output_device() { new_ucmd!() .args(&["-o", "/dev/null"]) @@ -1933,6 +2020,7 @@ fn test_wrong_args_exit_code() { #[test] #[cfg(unix)] +#[cfg_attr(wasi_runner, ignore = "WASI: no pipe/signal support")] fn test_tmp_files_deleted_on_sigint() { use rand::{RngExt as _, SeedableRng, rngs::SmallRng}; use rustix::process::{Pid, Signal, kill_process}; @@ -2130,6 +2218,10 @@ fn test_files0_from_non_utf8_filename() { #[test] #[cfg(unix)] +#[cfg_attr( + wasi_runner, + ignore = "WASI: opening a directory as a file reports EBADF instead of EISDIR" +)] fn test_files0_from_unreadable_source() { new_ucmd!() .args(&["--files0-from", "."]) @@ -2139,6 +2231,7 @@ fn test_files0_from_unreadable_source() { #[cfg(unix)] #[test] +#[cfg_attr(wasi_runner, ignore = "WASI sandbox: host paths not visible")] // Test files0-from with non-regular empty file fn test_files0_from_dev_null_is_empty() { new_ucmd!() @@ -2226,6 +2319,7 @@ fn test_files0_from_two_entries_trailing_nul() { #[test] // Test files0-from with non-UTF-8 filenames #[cfg(all(unix, not(target_vendor = "apple")))] +#[cfg_attr(wasi_runner, ignore = "WASI: argv/filenames must be valid UTF-8")] fn test_files0_from_non_utf8_content() { use std::os::unix::ffi::OsStringExt; let (at, mut ucmd) = at_and_ucmd!(); @@ -3212,6 +3306,7 @@ fn test_locale_collation_utf8() { } #[test] +#[cfg_attr(wasi_runner, ignore = "WASI: no locale data")] fn test_locale_interleaved_en_us_utf8() { // Test case for issue: locale-based collation support // In en_US.UTF-8, lowercase and uppercase letters should interleave @@ -3284,6 +3379,7 @@ fn test_locale_with_ignore_case_flag() { } #[test] +#[cfg_attr(wasi_runner, ignore = "WASI: no locale data")] fn test_locale_complex_utf8_sorting() { // More complex test with mixed case and special characters // In en_US.UTF-8, should respect locale collation rules @@ -3308,6 +3404,7 @@ fn test_locale_posix_sort_debug_message() { } #[test] +#[cfg_attr(wasi_runner, ignore = "WASI: no locale data")] fn test_locale_utf8_sort_debug_message() { new_ucmd!() .env("LC_ALL", "en_US.UTF-8") @@ -3319,6 +3416,7 @@ fn test_locale_utf8_sort_debug_message() { #[test] #[cfg(unix)] +#[cfg_attr(wasi_runner, ignore = "WASI: no locale data")] fn test_failed_to_set_locale_debug_message() { let result = new_ucmd!() .env("LC_ALL", "not-valid-locale") @@ -3350,6 +3448,7 @@ e f 5436 down data path1 path2 path3 path4 path5\n"; } #[test] +#[cfg_attr(wasi_runner, ignore = "WASI: no locale data")] fn test_consistent_sorting_with_i18n_collate() { // Regression test for issue #11980 // Lexicographic fallback sorting for equal sorting keys for 01 and 0_1