Skip to content

Commit 362b58c

Browse files
committed
fix(wasi): sort ordering check and external-merge separator
sort's ordering check runs single-threaded on WASI, which has no thread support, reading every chunk up front instead of streaming from a background reader thread. Its external-merge path also inserts a missing separator between concatenated files so lines don't merge across file boundaries.
1 parent d99a177 commit 362b58c

2 files changed

Lines changed: 127 additions & 9 deletions

File tree

src/uu/sort/src/check.rs

Lines changed: 120 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -11,21 +11,88 @@ use crate::{
1111
compare_by, open,
1212
};
1313
use itertools::Itertools;
14+
use std::{cmp::Ordering, ffi::OsStr};
15+
#[cfg(not(target_os = "wasi"))]
1416
use std::{
15-
cmp::Ordering,
16-
ffi::OsStr,
1717
io::Read,
1818
iter,
1919
sync::mpsc::{Receiver, SyncSender, sync_channel},
2020
thread,
2121
};
2222
use uucore::error::UResult;
2323

24+
fn buffer_size(settings: &GlobalSettings) -> usize {
25+
if settings.buffer_size < 100 * 1024 {
26+
// when the buffer size is smaller than 100KiB we choose it instead of the default.
27+
// this improves testability.
28+
settings.buffer_size
29+
} else {
30+
100 * 1024
31+
}
32+
}
33+
34+
/// Given the chunks of a file (in order), find the first pair of adjacent
35+
/// lines that violates the requested ordering and report it as a
36+
/// [`SortError::Disorder`].
37+
#[cfg(target_os = "wasi")]
38+
fn check_chunks(
39+
path: &OsStr,
40+
settings: &GlobalSettings,
41+
max_allowed_cmp: Ordering,
42+
chunks: impl Iterator<Item = Chunk>,
43+
) -> UResult<()> {
44+
let mut prev_chunk: Option<Chunk> = None;
45+
let mut line_idx = 0;
46+
for chunk in chunks {
47+
line_idx += 1;
48+
if let Some(prev_chunk) = &prev_chunk {
49+
// Check if the first element of the new chunk is greater than the last
50+
// element from the previous chunk
51+
let prev_last = prev_chunk.lines().last().unwrap();
52+
let new_first = chunk.lines().first().unwrap();
53+
54+
if compare_by(
55+
prev_last,
56+
new_first,
57+
settings,
58+
prev_chunk.line_data(),
59+
chunk.line_data(),
60+
) > max_allowed_cmp
61+
{
62+
return Err(SortError::Disorder {
63+
file: path.to_owned(),
64+
line_number: line_idx,
65+
line: String::from_utf8_lossy(new_first.line).into_owned(),
66+
silent: settings.check_silent,
67+
}
68+
.into());
69+
}
70+
}
71+
72+
for (a, b) in chunk.lines().iter().tuple_windows() {
73+
line_idx += 1;
74+
if compare_by(a, b, settings, chunk.line_data(), chunk.line_data()) > max_allowed_cmp {
75+
return Err(SortError::Disorder {
76+
file: path.to_owned(),
77+
line_number: line_idx,
78+
line: String::from_utf8_lossy(b.line).into_owned(),
79+
silent: settings.check_silent,
80+
}
81+
.into());
82+
}
83+
}
84+
85+
prev_chunk = Some(chunk);
86+
}
87+
Ok(())
88+
}
89+
2490
/// Check if the file at `path` is ordered.
2591
///
2692
/// # Returns
2793
///
2894
/// The code we should exit with.
95+
#[cfg(not(target_os = "wasi"))]
2996
pub fn check(path: &OsStr, settings: &GlobalSettings) -> UResult<()> {
3097
let max_allowed_cmp = if settings.unique {
3198
// If `unique` is enabled, the previous line must compare _less_ to the next one.
@@ -42,13 +109,7 @@ pub fn check(path: &OsStr, settings: &GlobalSettings) -> UResult<()> {
42109
move || reader(file, &recycled_receiver, &loaded_sender, &settings)
43110
});
44111
for _ in 0..2 {
45-
let _ = recycled_sender.send(RecycledChunk::new(if settings.buffer_size < 100 * 1024 {
46-
// when the buffer size is smaller than 100KiB we choose it instead of the default.
47-
// this improves testability.
48-
settings.buffer_size
49-
} else {
50-
100 * 1024
51-
}));
112+
let _ = recycled_sender.send(RecycledChunk::new(buffer_size(settings)));
52113
}
53114

54115
let mut prev_chunk: Option<Chunk> = None;
@@ -114,7 +175,27 @@ pub fn check(path: &OsStr, settings: &GlobalSettings) -> UResult<()> {
114175
result
115176
}
116177

178+
/// Check if the file at `path` is ordered.
179+
///
180+
/// WASI has no thread support, so this reads every chunk up front on the
181+
/// current thread instead of streaming them from a background reader thread.
182+
///
183+
/// # Returns
184+
///
185+
/// The code we should exit with.
186+
#[cfg(target_os = "wasi")]
187+
pub fn check(path: &OsStr, settings: &GlobalSettings) -> UResult<()> {
188+
let max_allowed_cmp = if settings.unique {
189+
Ordering::Less
190+
} else {
191+
Ordering::Equal
192+
};
193+
let chunks = read_all_chunks(path, settings)?;
194+
check_chunks(path, settings, max_allowed_cmp, chunks.into_iter())
195+
}
196+
117197
/// The function running on the reader thread.
198+
#[cfg(not(target_os = "wasi"))]
118199
fn reader(
119200
mut file: Box<dyn Read + Send>,
120201
receiver: &Receiver<RecycledChunk>,
@@ -139,3 +220,33 @@ fn reader(
139220
}
140221
Ok(())
141222
}
223+
224+
/// Read every chunk of `path` up front, without any recycling or background
225+
/// thread. Used on WASI, which has no thread support.
226+
#[cfg(target_os = "wasi")]
227+
fn read_all_chunks(path: &OsStr, settings: &GlobalSettings) -> UResult<Vec<Chunk>> {
228+
let mut file = open(path)?;
229+
let mut carry_over = vec![];
230+
let mut chunks = Vec::new();
231+
let (sender, receiver) = std::sync::mpsc::sync_channel(1);
232+
loop {
233+
let recycled = RecycledChunk::new(buffer_size(settings));
234+
let should_continue = chunks::read(
235+
&sender,
236+
recycled,
237+
None,
238+
&mut carry_over,
239+
&mut file,
240+
&mut std::iter::empty(),
241+
settings.line_ending.into(),
242+
settings,
243+
)?;
244+
while let Ok(chunk) = receiver.try_recv() {
245+
chunks.push(chunk);
246+
}
247+
if !should_continue {
248+
break;
249+
}
250+
}
251+
Ok(chunks)
252+
}

src/uu/sort/src/ext_sort/wasi.rs

Lines changed: 7 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -31,6 +31,13 @@ pub fn ext_sort(
3131
// moderately sized inputs; very large files may cause OOM.
3232
let mut input = Vec::new();
3333
for file in files {
34+
// Insert the separator between files whose preceding content doesn't
35+
// already end with one; otherwise the last line of one file would
36+
// merge with the first line of the next, e.g. "a\nb" + "b" ->
37+
// "a\nbb" instead of "a\nb" + '\n' + "b".
38+
if !input.is_empty() && input.last() != Some(&separator) {
39+
input.push(separator);
40+
}
3441
file?.read_to_end(&mut input)?;
3542
}
3643
if input.is_empty() {

0 commit comments

Comments
 (0)