sort: add complete no-thread WASI fallback

.github/workflows/wasi.yml

@@ -36,6 +36,11 @@ jobs:
     - uses: Swatinem/rust-cache@v2
       with:
         key: "${{ matrix.job.target }}"
+    - 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
@@ -63,7 +68,7 @@ jobs:
         #   arch b2sum cat cksum cp 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 tail touch tsort uname uniq
+        #   sha512sum shred sleep split tail touch tsort uname uniq
         #   vdir
         UUTESTS_BINARY_PATH="$(pwd)/target/${{ matrix.job.target }}/debug/coreutils.wasm" \
         UUTESTS_WASM_RUNNER=wasmtime \
@@ -72,6 +77,7 @@ jobs:
           test_comm:: test_cut:: test_dirname:: test_echo:: \
           test_expand:: 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_od:: test_paste:: test_printf:: test_shuf:: test_sort:: \
+          test_sum:: \
           test_tee:: test_tr:: test_true:: test_truncate:: \
           test_unexpand:: test_unlink:: test_wc:: test_yes::

src/uu/sort/Cargo.toml

@@ -29,7 +29,6 @@ compare = { workspace = true }
 itertools = { workspace = true }
 memchr = { workspace = true }
 rand = { workspace = true }
-rayon = { workspace = true }
 self_cell = { workspace = true }
 tempfile = { workspace = true }
 thiserror = { workspace = true }
@@ -47,6 +46,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 }
 

src/uu/sort/build.rs

@@ -0,0 +1,12 @@
+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();
+
+    if target_os == "wasi" && target != "wasm32-wasip1-threads" {
+        println!("cargo::rustc-cfg=wasi_no_threads");
+    }
+}

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
+//! atomics, 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)
+}

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<dyn Read + Send>,
+    chunk_size: usize,
+) -> UResult<()> {
+    let separator = settings.line_ending.into();
+    let mut carry_over = vec![];
+    let mut prev_chunk: Option<Chunk> = None;
+    let mut spare_recycled: Option<RecycledChunk> = 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(())
+}

src/uu/sort/src/check.rs → src/uu/sort/src/check/threaded.rs

@@ -3,67 +3,50 @@
 // 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<dyn Read + Send>,
+    chunk_size: usize,
+) -> UResult<()> {
     let (recycled_sender, recycled_receiver) = sync_channel(2);
     let (loaded_sender, loaded_receiver) = sync_channel(2);
     thread::spawn({
         let settings = settings.clone();
         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<Chunk> = 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() {}