Merge pull request #2443 from miDeb/sort/data-oriented

sort: separate additional data from the Line struct

src/uu/sort/src/check.rs

@@ -8,7 +8,7 @@
 //! Check if a file is ordered
 
 use crate::{
-    chunks::{self, Chunk},
+    chunks::{self, Chunk, RecycledChunk},
     compare_by, open, GlobalSettings,
 };
 use itertools::Itertools;
@@ -34,7 +34,7 @@ pub fn check(path: &str, settings: &GlobalSettings) -> i32 {
         move || reader(file, recycled_receiver, loaded_sender, &settings)
     });
     for _ in 0..2 {
-        let _ = recycled_sender.send(Chunk::new(vec![0; 100 * 1024], |_| Vec::new()));
+        let _ = recycled_sender.send(RecycledChunk::new(100 * 1024));
     }
 
     let mut prev_chunk: Option<Chunk> = None;
@@ -44,21 +44,29 @@ pub fn check(path: &str, settings: &GlobalSettings) -> i32 {
         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.borrow_lines().last().unwrap();
-            let new_first = chunk.borrow_lines().first().unwrap();
+            let prev_last = prev_chunk.lines().last().unwrap();
+            let new_first = chunk.lines().first().unwrap();
 
-            if compare_by(prev_last, new_first, settings) == Ordering::Greater {
+            if compare_by(
+                prev_last,
+                new_first,
+                settings,
+                prev_chunk.line_data(),
+                chunk.line_data(),
+            ) == Ordering::Greater
+            {
                 if !settings.check_silent {
                     println!("sort: {}:{}: disorder: {}", path, line_idx, new_first.line);
                 }
                 return 1;
             }
-            let _ = recycled_sender.send(prev_chunk);
+            let _ = recycled_sender.send(prev_chunk.recycle());
         }
 
-        for (a, b) in chunk.borrow_lines().iter().tuple_windows() {
+        for (a, b) in chunk.lines().iter().tuple_windows() {
             line_idx += 1;
-            if compare_by(a, b, settings) == Ordering::Greater {
+            if compare_by(a, b, settings, chunk.line_data(), chunk.line_data()) == Ordering::Greater
+            {
                 if !settings.check_silent {
                     println!("sort: {}:{}: disorder: {}", path, line_idx, b.line);
                 }
@@ -74,16 +82,15 @@ pub fn check(path: &str, settings: &GlobalSettings) -> i32 {
 /// The function running on the reader thread.
 fn reader(
     mut file: Box<dyn Read + Send>,
-    receiver: Receiver<Chunk>,
+    receiver: Receiver<RecycledChunk>,
     sender: SyncSender<Chunk>,
     settings: &GlobalSettings,
 ) {
     let mut carry_over = vec![];
-    for chunk in receiver.iter() {
-        let (recycled_lines, recycled_buffer) = chunk.recycle();
+    for recycled_chunk in receiver.iter() {
         let should_continue = chunks::read(
             &sender,
-            recycled_buffer,
+            recycled_chunk,
             None,
             &mut carry_over,
             &mut file,
@@ -93,7 +100,6 @@ fn reader(
             } else {
                 b'\n'
             },
-            recycled_lines,
             settings,
         );
         if !should_continue {

src/uu/sort/src/chunks.rs

@@ -15,7 +15,7 @@ use std::{
 use memchr::memchr_iter;
 use ouroboros::self_referencing;
 
-use crate::{GlobalSettings, Line};
+use crate::{numeric_str_cmp::NumInfo, GeneralF64ParseResult, GlobalSettings, Line};
 
 /// The chunk that is passed around between threads.
 /// `lines` consist of slices into `buffer`.
@@ -25,28 +25,87 @@ pub struct Chunk {
     pub buffer: Vec<u8>,
     #[borrows(buffer)]
     #[covariant]
-    pub lines: Vec<Line<'this>>,
+    pub contents: ChunkContents<'this>,
+}
+
+#[derive(Debug)]
+pub struct ChunkContents<'a> {
+    pub lines: Vec<Line<'a>>,
+    pub line_data: LineData<'a>,
+}
+
+#[derive(Debug)]
+pub struct LineData<'a> {
+    pub selections: Vec<&'a str>,
+    pub num_infos: Vec<NumInfo>,
+    pub parsed_floats: Vec<GeneralF64ParseResult>,
 }
 
 impl Chunk {
     /// Destroy this chunk and return its components to be reused.
-    ///
-    /// # Returns
-    ///
-    /// * The `lines` vector, emptied
-    /// * The `buffer` vector, **not** emptied
-    pub fn recycle(mut self) -> (Vec<Line<'static>>, Vec<u8>) {
-        let recycled_lines = self.with_lines_mut(|lines| {
-            lines.clear();
-            unsafe {
+    pub fn recycle(mut self) -> RecycledChunk {
+        let recycled_contents = self.with_contents_mut(|contents| {
+            contents.lines.clear();
+            contents.line_data.selections.clear();
+            contents.line_data.num_infos.clear();
+            contents.line_data.parsed_floats.clear();
+            let lines = unsafe {
                 // SAFETY: It is safe to (temporarily) transmute to a vector of lines with a longer lifetime,
                 // because the vector is empty.
                 // Transmuting is necessary to make recycling possible. See https://github.com/rust-lang/rfcs/pull/2802
                 // for a rfc to make this unnecessary. Its example is similar to the code here.
-                std::mem::transmute::<Vec<Line<'_>>, Vec<Line<'static>>>(std::mem::take(lines))
-            }
+                std::mem::transmute::<Vec<Line<'_>>, Vec<Line<'static>>>(std::mem::take(
+                    &mut contents.lines,
+                ))
+            };
+            let selections = unsafe {
+                // SAFETY: (same as above) It is safe to (temporarily) transmute to a vector of &str with a longer lifetime,
+                // because the vector is empty.
+                std::mem::transmute::<Vec<&'_ str>, Vec<&'static str>>(std::mem::take(
+                    &mut contents.line_data.selections,
+                ))
+            };
+            (
+                lines,
+                selections,
+                std::mem::take(&mut contents.line_data.num_infos),
+                std::mem::take(&mut contents.line_data.parsed_floats),
+            )
         });
-        (recycled_lines, self.into_heads().buffer)
+        RecycledChunk {
+            lines: recycled_contents.0,
+            selections: recycled_contents.1,
+            num_infos: recycled_contents.2,
+            parsed_floats: recycled_contents.3,
+            buffer: self.into_heads().buffer,
+        }
+    }
+
+    pub fn lines(&self) -> &Vec<Line> {
+        &self.borrow_contents().lines
+    }
+    pub fn line_data(&self) -> &LineData {
+        &self.borrow_contents().line_data
+    }
+}
+
+pub struct RecycledChunk {
+    lines: Vec<Line<'static>>,
+    selections: Vec<&'static str>,
+    num_infos: Vec<NumInfo>,
+    parsed_floats: Vec<GeneralF64ParseResult>,
+    buffer: Vec<u8>,
+}
+
+impl RecycledChunk {
+    pub fn new(capacity: usize) -> Self {
+        RecycledChunk {
+            lines: Vec::new(),
+            selections: Vec::new(),
+            num_infos: Vec::new(),
+            parsed_floats: Vec::new(),
+            buffer: vec![0; capacity],
+        }
     }
 }
 
@@ -63,28 +122,32 @@ impl Chunk {
 /// (see also `read_to_chunk` for a more detailed documentation)
 ///
 /// * `sender`: The sender to send the lines to the sorter.
-/// * `buffer`: The recycled buffer. All contents will be overwritten, but it must already be filled.
+/// * `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.
-/// * `lines`: The recycled vector to fill with lines. Must be empty.
 /// * `settings`: The global settings.
 #[allow(clippy::too_many_arguments)]
 pub fn read<T: Read>(
     sender: &SyncSender<Chunk>,
-    mut buffer: Vec<u8>,
+    recycled_chunk: RecycledChunk,
     max_buffer_size: Option<usize>,
     carry_over: &mut Vec<u8>,
     file: &mut T,
     next_files: &mut impl Iterator<Item = T>,
     separator: u8,
-    lines: Vec<Line<'static>>,
     settings: &GlobalSettings,
 ) -> bool {
-    assert!(lines.is_empty());
+    let RecycledChunk {
+        lines,
+        selections,
+        num_infos,
+        parsed_floats,
+        mut buffer,
+    } = recycled_chunk;
     if buffer.len() < carry_over.len() {
         buffer.resize(carry_over.len() + 10 * 1024, 0);
     }
@@ -101,15 +164,25 @@ pub fn read<T: Read>(
     carry_over.extend_from_slice(&buffer[read..]);
 
     if read != 0 {
-        let payload = Chunk::new(buffer, |buf| {
+        let payload = Chunk::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<Line<'static>> to make recycling possible.
+                std::mem::transmute::<Vec<&'static str>, Vec<&'_ str>>(selections)
+            };
             let mut lines = unsafe {
-                // SAFETY: It is safe to transmute to a vector of lines with shorter lifetime,
+                // SAFETY: (same as above) It is safe to transmute to a vector of lines with shorter lifetime,
                 // because it was only temporarily transmuted to a Vec<Line<'static>> to make recycling possible.
                 std::mem::transmute::<Vec<Line<'static>>, Vec<Line<'_>>>(lines)
             };
-            let read = crash_if_err!(1, std::str::from_utf8(&buf[..read]));
-            parse_lines(read, &mut lines, separator, settings);
-            lines
+            let read = crash_if_err!(1, std::str::from_utf8(&buffer[..read]));
+            let mut line_data = LineData {
+                selections,
+                num_infos,
+                parsed_floats,
+            };
+            parse_lines(read, &mut lines, &mut line_data, separator, settings);
+            ChunkContents { lines, line_data }
         });
         sender.send(payload).unwrap();
     }
@@ -120,6 +193,7 @@ pub fn read<T: Read>(
 fn parse_lines<'a>(
     mut read: &'a str,
     lines: &mut Vec<Line<'a>>,
+    line_data: &mut LineData<'a>,
     separator: u8,
     settings: &GlobalSettings,
 ) {
@@ -128,9 +202,15 @@ fn parse_lines<'a>(
         read = &read[..read.len() - 1];
     }
 
+    assert!(lines.is_empty());
+    assert!(line_data.selections.is_empty());
+    assert!(line_data.num_infos.is_empty());
+    assert!(line_data.parsed_floats.is_empty());
+    let mut token_buffer = vec![];
     lines.extend(
         read.split(separator as char)
-            .map(|line| Line::create(line, settings)),
+            .enumerate()
+            .map(|(index, line)| Line::create(line, index, line_data, &mut token_buffer, settings)),
     );
 }
 

src/uu/sort/src/ext_sort.rs

@@ -23,15 +23,16 @@ use std::{
 
 use itertools::Itertools;
 
+use crate::chunks::RecycledChunk;
 use crate::merge::ClosedTmpFile;
 use crate::merge::WriteableCompressedTmpFile;
 use crate::merge::WriteablePlainTmpFile;
 use crate::merge::WriteableTmpFile;
-use crate::Line;
 use crate::{
     chunks::{self, Chunk},
-    compare_by, merge, output_sorted_lines, sort_by, GlobalSettings,
+    compare_by, merge, sort_by, GlobalSettings,
 };
+use crate::{print_sorted, Line};
 use tempfile::TempDir;
 
 const START_BUFFER_SIZE: usize = 8_000;
@@ -98,16 +99,39 @@ fn reader_writer<F: Iterator<Item = Box<dyn Read + Send>>, Tmp: WriteableTmpFile
             merger.write_all(settings);
         }
         ReadResult::SortedSingleChunk(chunk) => {
-            output_sorted_lines(chunk.borrow_lines().iter(), settings);
+            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,
+                );
+            } else {
+                print_sorted(chunk.lines().iter(), settings);
+            }
         }
         ReadResult::SortedTwoChunks([a, b]) => {
-            let merged_iter = a
-                .borrow_lines()
-                .iter()
-                .merge_by(b.borrow_lines().iter(), |line_a, line_b| {
-                    compare_by(line_a, line_b, settings) != Ordering::Greater
-                });
-            output_sorted_lines(merged_iter, settings);
+            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,
+                );
+            } else {
+                print_sorted(merged_iter.map(|(line, _)| line), settings);
+            }
         }
         ReadResult::EmptyInput => {
             // don't output anything
@@ -118,7 +142,9 @@ fn reader_writer<F: Iterator<Item = Box<dyn Read + Send>>, Tmp: WriteableTmpFile
 /// The function that is executed on the sorter thread.
 fn sorter(receiver: Receiver<Chunk>, sender: SyncSender<Chunk>, settings: GlobalSettings) {
     while let Ok(mut payload) = receiver.recv() {
-        payload.with_lines_mut(|lines| sort_by(lines, &settings));
+        payload.with_contents_mut(|contents| {
+            sort_by(&mut contents.lines, &settings, &contents.line_data)
+        });
         sender.send(payload).unwrap();
     }
 }
@@ -154,20 +180,16 @@ fn read_write_loop<I: WriteableTmpFile>(
     for _ in 0..2 {
         let should_continue = chunks::read(
             &sender,
-            vec![
-                0;
-                if START_BUFFER_SIZE < buffer_size {
-                    START_BUFFER_SIZE
-                } else {
-                    buffer_size
-                }
-            ],
+            RecycledChunk::new(if START_BUFFER_SIZE < buffer_size {
+                START_BUFFER_SIZE
+            } else {
+                buffer_size
+            }),
             Some(buffer_size),
             &mut carry_over,
             &mut file,
             &mut files,
             separator,
-            Vec::new(),
             settings,
         );
 
@@ -216,18 +238,17 @@ fn read_write_loop<I: WriteableTmpFile>(
 
         file_number += 1;
 
-        let (recycled_lines, recycled_buffer) = chunk.recycle();
+        let recycled_chunk = chunk.recycle();
 
         if let Some(sender) = &sender_option {
             let should_continue = chunks::read(
                 sender,
-                recycled_buffer,
+                recycled_chunk,
                 None,
                 &mut carry_over,
                 &mut file,
                 &mut files,
                 separator,
-                recycled_lines,
                 settings,
             );
             if !should_continue {
@@ -245,12 +266,9 @@ fn write<I: WriteableTmpFile>(
     compress_prog: Option<&str>,
     separator: u8,
 ) -> I::Closed {
-    chunk.with_lines_mut(|lines| {
-        // Write the lines to the file
-        let mut tmp_file = I::create(file, compress_prog);
-        write_lines(lines, tmp_file.as_write(), separator);
-        tmp_file.finished_writing()
-    })
+    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<'a, T: Write>(lines: &[Line<'a>], writer: &mut T, separator: u8) {

src/uu/sort/src/merge.rs

@@ -24,7 +24,7 @@ use itertools::Itertools;
 use tempfile::TempDir;
 
 use crate::{
-    chunks::{self, Chunk},
+    chunks::{self, Chunk, RecycledChunk},
     compare_by, GlobalSettings,
 };
 
@@ -125,14 +125,14 @@ fn merge_without_limit<M: MergeInput + 'static, F: Iterator<Item = M>>(
         }));
         // Send the initial chunk to trigger a read for each file
         request_sender
-            .send((file_number, Chunk::new(vec![0; 8 * 1024], |_| Vec::new())))
+            .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, Chunk::new(vec![0; 8 * 1024], |_| Vec::new())))
+            .send((file_number, RecycledChunk::new(8 * 1024)))
             .unwrap();
     }
 
@@ -181,13 +181,12 @@ struct ReaderFile<M: MergeInput> {
 
 /// The function running on the reader thread.
 fn reader(
-    recycled_receiver: Receiver<(usize, Chunk)>,
+    recycled_receiver: Receiver<(usize, RecycledChunk)>,
     files: &mut [Option<ReaderFile<impl MergeInput>>],
     settings: &GlobalSettings,
     separator: u8,
 ) {
-    for (file_idx, chunk) in recycled_receiver.iter() {
-        let (recycled_lines, recycled_buffer) = chunk.recycle();
+    for (file_idx, recycled_chunk) in recycled_receiver.iter() {
         if let Some(ReaderFile {
             file,
             sender,
@@ -196,13 +195,12 @@ fn reader(
         {
             let should_continue = chunks::read(
                 sender,
-                recycled_buffer,
+                recycled_chunk,
                 None,
                 carry_over,
                 file.as_read(),
                 &mut iter::empty(),
                 separator,
-                recycled_lines,
                 settings,
             );
             if !should_continue {
@@ -234,7 +232,7 @@ struct PreviousLine {
 /// Merges files together. This is **not** an iterator because of lifetime problems.
 pub struct FileMerger<'a> {
     heap: binary_heap_plus::BinaryHeap<MergeableFile, FileComparator<'a>>,
-    request_sender: Sender<(usize, Chunk)>,
+    request_sender: Sender<(usize, RecycledChunk)>,
     prev: Option<PreviousLine>,
 }
 
@@ -257,14 +255,16 @@ impl<'a> FileMerger<'a> {
                 file_number: file.file_number,
             });
 
-            file.current_chunk.with_lines(|lines| {
-                let current_line = &lines[file.line_idx];
+            file.current_chunk.with_contents(|contents| {
+                let current_line = &contents.lines[file.line_idx];
                 if settings.unique {
                     if let Some(prev) = &prev {
                         let cmp = compare_by(
-                            &prev.chunk.borrow_lines()[prev.line_idx],
+                            &prev.chunk.lines()[prev.line_idx],
                             current_line,
                             settings,
+                            prev.chunk.line_data(),
+                            file.current_chunk.line_data(),
                         );
                         if cmp == Ordering::Equal {
                             return;
@@ -274,8 +274,7 @@ impl<'a> FileMerger<'a> {
                 current_line.print(out, settings);
             });
 
-            let was_last_line_for_file =
-                file.current_chunk.borrow_lines().len() == file.line_idx + 1;
+            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() {
@@ -295,7 +294,7 @@ impl<'a> FileMerger<'a> {
                     // 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))
+                        .send((prev.file_number, prev_chunk.recycle()))
                         .ok();
                 }
             }
@@ -312,9 +311,11 @@ struct FileComparator<'a> {
 impl<'a> Compare<MergeableFile> for FileComparator<'a> {
     fn compare(&self, a: &MergeableFile, b: &MergeableFile) -> Ordering {
         let mut cmp = compare_by(
-            &a.current_chunk.borrow_lines()[a.line_idx],
-            &b.current_chunk.borrow_lines()[b.line_idx],
+            &a.current_chunk.lines()[a.line_idx],
+            &b.current_chunk.lines()[b.line_idx],
             self.settings,
+            a.current_chunk.line_data(),
+            b.current_chunk.line_data(),
         );
         if cmp == Ordering::Equal {
             // To make sorting stable, we need to consider the file number as well,