du: start printing output immediately

src/uu/du/src/du.rs

@@ -27,6 +27,8 @@
 use std::path::Path;
 use std::path::PathBuf;
 use std::str::FromStr;
+use std::sync::mpsc;
+use std::thread;
 use std::time::{Duration, UNIX_EPOCH};
 use std::{error::Error, fmt::Display};
 use uucore::display::{print_verbatim, Quotable};
@@ -81,6 +83,7 @@
 // TODO: Support Z & Y (currently limited by size of u64)
 const UNITS: [(char, u32); 6] = [('E', 6), ('P', 5), ('T', 4), ('G', 3), ('M', 2), ('K', 1)];
 
+#[derive(Clone)]
 struct Options {
     all: bool,
     max_depth: Option<usize>,
@@ -93,7 +96,7 @@
     verbose: bool,
 }
 
-#[derive(PartialEq)]
+#[derive(PartialEq, Clone)]
 enum Deref {
     All,
     Args(Vec<PathBuf>),
@@ -106,6 +109,7 @@
     dev_id: u64,
 }
 
+#[derive(Clone)]
 struct Stat {
     path: PathBuf,
     is_dir: bool,
@@ -289,6 +293,82 @@
     }
 }
 
+struct StatPrinter {
+    matches: ArgMatches,
+    threshold: Option<Threshold>,
+    summarize: bool,
+    time_format_str: String,
+    line_ending: LineEnding,
+    options: Options,
+}
+
+impl StatPrinter {
+    fn new(
+        matches: ArgMatches,
+        threshold: Option<Threshold>,
+        summarize: bool,
+        time_format_str: String,
+        line_ending: LineEnding,
+        options: Options,
+    ) -> StatPrinter {
+        StatPrinter {
+            matches,
+            threshold,
+            summarize,
+            time_format_str,
+            line_ending,
+            options,
+        }
+    }
+
+    fn print(
+        &self,
+        stat: &Stat,
+        depth: usize,
+        convert_size: &dyn Fn(u64) -> String,
+    ) -> UResult<()> {
+        let size = choose_size(&self.matches, &stat);
+
+        if !self
+            .threshold
+            .map_or(false, |threshold| threshold.should_exclude(size))
+            && self
+                .options
+                .max_depth
+                .map_or(true, |max_depth| depth <= max_depth)
+        {
+            if self.matches.contains_id(options::TIME) {
+                if !self.summarize || depth == 0 {
+                    let tm = {
+                        let secs = self
+                            .matches
+                            .get_one::<String>(options::TIME)
+                            .map(|s| get_time_secs(s, &stat))
+                            .transpose()?
+                            .unwrap_or(stat.modified);
+                        DateTime::<Local>::from(UNIX_EPOCH + Duration::from_secs(secs))
+                    };
+                    let time_str = tm.format(&self.time_format_str).to_string();
+                    print!("{}\t{}\t", convert_size(size), time_str);
+                    print_verbatim(&stat.path).unwrap();
+                    print!("{}", self.line_ending);
+                }
+            } else if !self.summarize || depth == 0 {
+                print!("{}\t", convert_size(size));
+                print_verbatim(&stat.path).unwrap();
+                print!("{}", self.line_ending);
+            }
+        }
+
+        Ok(())
+    }
+}
+
+struct StatPrintInfo {
+    stat: Stat,
+    depth: usize,
+}
+
 // this takes `my_stat` to avoid having to stat files multiple times.
 // XXX: this should use the impl Trait return type when it is stabilized
 #[allow(clippy::cognitive_complexity)]
@@ -298,7 +378,8 @@
     depth: usize,
     seen_inodes: &mut HashSet<FileInfo>,
     exclude: &[Pattern],
-) -> Box<dyn DoubleEndedIterator<Item = Stat>> {
+    tx: &mpsc::Sender<StatPrintInfo>,
+) -> UResult<Box<dyn DoubleEndedIterator<Item = Stat>>> {
     let mut stats = vec![];
     let mut futures = vec![];
 
@@ -309,7 +390,12 @@
                 show!(
                     e.map_err_context(|| format!("cannot read directory {}", my_stat.path.quote()))
                 );
-                return Box::new(iter::once(my_stat));
+                tx.send(StatPrintInfo {
+                    stat: my_stat.clone(),
+                    depth,
+                })
+                .unwrap();
+                return Ok(Box::new(iter::once(my_stat)));
             }
         };
 
@@ -360,12 +446,18 @@
                                     depth + 1,
                                     seen_inodes,
                                     exclude,
-                                ));
+                                    tx,
+                                )?);
                             } else {
                                 my_stat.size += this_stat.size;
                                 my_stat.blocks += this_stat.blocks;
                                 my_stat.inodes += 1;
                                 if options.all {
+                                    tx.send(StatPrintInfo {
+                                        stat: this_stat.clone(),
+                                        depth,
+                                    })
+                                    .unwrap();
                                     stats.push(this_stat);
                                 }
                             }
@@ -390,8 +482,16 @@
             .max_depth
             .map_or(true, |max_depth| depth < max_depth)
     }));
+
+    tx.send(StatPrintInfo {
+        stat: my_stat.clone(),
+        depth,
+    })
+    .unwrap();
+
     stats.push(my_stat);
-    Box::new(stats.into_iter())
+
+    Ok(Box::new(stats.into_iter()))
 }
 
 fn convert_size_human(size: u64, multiplier: u64, _block_size: u64) -> String {
@@ -622,6 +722,44 @@
 
     let line_ending = LineEnding::from_zero_flag(matches.get_flag(options::NULL));
 
+    let stat_printer = StatPrinter::new(
+        matches.clone(),
+        threshold.clone(),
+        summarize,
+        time_format_str.to_owned(),
+        line_ending,
+        options.clone(),
+    );
+
+    let (tx, rx) = mpsc::channel::<StatPrintInfo>();
+
+    let printing_thread = thread::spawn({
+        let matchesx = matches.clone();
+        let optionsx = options.clone();
+        move || {
+            let convert_size_fn = get_convert_size_fn(&matchesx);
+
+            let convert_size = |size: u64| {
+                if optionsx.inodes {
+                    size.to_string()
+                } else {
+                    convert_size_fn(size, multiplier, block_size)
+                }
+            };
+            loop {
+                let stat_info = rx.recv();
+
+                match stat_info {
+                    // TODO: actually use depth properly
+                    Ok(stat_info) => stat_printer
+                        .print(&stat_info.stat, stat_info.depth, &convert_size)
+                        .unwrap(),
+                    Err(_) => break,
+                }
+            }
+        }
+    });
+
     let excludes = build_exclude_patterns(&matches)?;
 
     let mut grand_total = 0;
@@ -647,43 +785,11 @@
             if let Some(inode) = stat.inode {
                 seen_inodes.insert(inode);
             }
-            let iter = du(stat, &options, 0, &mut seen_inodes, &excludes);
+            let iter = du(stat, &options, 0, &mut seen_inodes, &excludes, &tx)?;
 
             // Sum up all the returned `Stat`s and display results
-            let (_, len) = iter.size_hint();
-            let len = len.unwrap();
-            for (index, stat) in iter.enumerate() {
-                let size = choose_size(&matches, &stat);
-
-                if threshold.map_or(false, |threshold| threshold.should_exclude(size)) {
-                    continue;
-                }
-
-                if matches.contains_id(options::TIME) {
-                    let tm = {
-                        let secs = matches
-                            .get_one::<String>(options::TIME)
-                            .map(|s| get_time_secs(s, &stat))
-                            .transpose()?
-                            .unwrap_or(stat.modified);
-                        DateTime::<Local>::from(UNIX_EPOCH + Duration::from_secs(secs))
-                    };
-                    if !summarize || index == len - 1 {
-                        let time_str = tm.format(time_format_str).to_string();
-                        print!("{}\t{}\t", convert_size(size), time_str);
-                        print_verbatim(stat.path).unwrap();
-                        print!("{line_ending}");
-                    }
-                } else if !summarize || index == len - 1 {
-                    print!("{}\t", convert_size(size));
-                    print_verbatim(stat.path).unwrap();
-                    print!("{line_ending}");
-                }
-                if options.total && index == (len - 1) {
-                    // The last element will be the total size of the the path under
-                    // path_string.  We add it to the grand total.
-                    grand_total += size;
-                }
+            if let Some(last_stat) = iter.last() {
+                grand_total += choose_size(&matches, &last_stat);
             }
         } else {
             show_error!(
@@ -695,6 +801,12 @@
         }
     }
 
+    drop(tx);
+
+    printing_thread
+        .join()
+        .expect("Failed to join with printing thread.");
+
     if options.total {
         print!("{}\ttotal", convert_size(grand_total));
         print!("{line_ending}");