multithreading improvements

src/main.rs

@@ -4,6 +4,7 @@ mod interactive;
 mod params;
 mod processor;
 mod scanner;
+mod server;
 
 use anyhow::Result;
 use clap::Parser;
@@ -12,19 +13,42 @@ use params::Params;
 use processor::Processor;
 use scanner::Scanner;
 
+use self::fileinfo::FileInfo;
+use std::sync::Arc;
+use std::sync::Mutex;
+use std::thread;
+
 fn main() -> Result<()> {
+    let files: Arc<Mutex<Vec<FileInfo>>> = Arc::new(Mutex::new(Vec::new()));
     let app_args = Params::parse();
-    let scan_results = Scanner::build(&app_args)?.scan()?;
-    let mut processor = Processor::new(scan_results);
 
-    processor.sizewise()?;
-    processor.hashwise()?;
+    // TODO: RACE CONDITION BETWEEN SCANNER AND PROCESSOR
+    let file_arc_clone = files.clone();
+    let app_args_clone = app_args.clone();
 
-    let results = processor.hashwise_results;
+    let scanner_thread = thread::spawn(move || {
+        Scanner::build(&app_args_clone)
+            .unwrap()
+            .scan(file_arc_clone)
+            .unwrap();
+    });
+
+    let file_arc_clone = files.clone();
+
+    let processor_thread = thread::spawn(move || {
+        let mut processor = Processor::new(file_arc_clone);
+        processor.sizewise().unwrap();
+        processor.hashwise().unwrap();
+
+        (processor.hashwise_results, processor.max_path_len)
+    });
+
+    scanner_thread.join().unwrap();
+    let results = processor_thread.join().unwrap();
 
     match app_args.interactive {
-        false => Formatter::print(results, processor.max_path_len, &app_args)?,
-        true => interactive::init(results, &app_args)?,
+        false => Formatter::print(results.0, results.1, &app_args)?,
+        true => interactive::init(results.0, &app_args)?,
     }
 
     Ok(())

src/processor.rs

@@ -2,20 +2,21 @@ use anyhow::Result;
 use dashmap::DashMap;
 use indicatif::{ParallelProgressIterator, ProgressBar, ProgressFinish, ProgressStyle};
 use rayon::prelude::{IntoParallelIterator, ParallelIterator};
+use std::sync::{Arc, Mutex};
 use std::{borrow::Cow, time::Duration};
 
 use crate::fileinfo::FileInfo;
 
 #[derive(Debug, Clone)]
 pub struct Processor {
-    pub files: Vec<FileInfo>,
+    pub files: Arc<Mutex<Vec<FileInfo>>>,
     pub hashwise_results: DashMap<String, Vec<FileInfo>>,
     pub sizewise_results: DashMap<u64, Vec<FileInfo>>,
     pub max_path_len: usize,
 }
 
 impl Processor {
-    pub fn new(files: Vec<FileInfo>) -> Self {
+    pub fn new(files: Arc<Mutex<Vec<FileInfo>>>) -> Self {
         Self {
             files,
             hashwise_results: DashMap::new(),
@@ -25,14 +26,19 @@ impl Processor {
     }
 
     pub fn hashwise(&mut self) -> Result<()> {
-        if self.sizewise_results.is_empty() {
+        let flist_size = {
+            let f = self.files.lock().unwrap();
+            f.len()
+        };
+        if flist_size < 1 {
             return Ok(());
         }
 
         let progress_style = ProgressStyle::with_template(
             "[{elapsed_precise}] {bar:40.cyan/blue} {pos:>7}/{len:7} {msg}",
         )?;
-        let progress_bar = ProgressBar::new(self.files.len() as u64);
+
+        let progress_bar = ProgressBar::new(flist_size as u64);
         progress_bar.set_style(progress_style);
         progress_bar.enable_steady_tick(Duration::from_millis(50));
         progress_bar.set_message("indexing file hashes");
@@ -72,26 +78,35 @@ impl Processor {
     }
 
     pub fn sizewise(&mut self) -> Result<()> {
-        if self.files.is_empty() {
+        let flist_size = {
+            let f = self.files.lock().unwrap();
+            f.len()
+        };
+
+        if flist_size < 1 {
             return Ok(());
         }
 
         let progress_style = ProgressStyle::with_template(
             "[{elapsed_precise}] {bar:40.cyan/blue} {pos:>7}/{len:7} {msg}",
         )?;
-        let progress_bar = ProgressBar::new(self.files.len() as u64);
+        let progress_bar = ProgressBar::new(flist_size as u64);
         progress_bar.set_style(progress_style);
         progress_bar.enable_steady_tick(Duration::from_millis(50));
         progress_bar.set_message("indexing file sizes");
 
-        self.files
-            .clone()
+        (0..flist_size)
             .into_par_iter()
             .progress_with(progress_bar)
             .with_finish(ProgressFinish::WithMessage(Cow::from(
                 "indexed files sizes",
             )))
-            .for_each(|file| {
+            .for_each(|findex| {
+                let file = {
+                    let files = self.files.lock().unwrap();
+                    files[findex].clone()
+                };
+
                 self.sizewise_results
                     .entry(file.size)
                     .and_modify(|fileset| fileset.push(file.clone()))

src/scanner.rs

@@ -2,6 +2,7 @@
 use crate::{fileinfo::FileInfo, params::Params};
 use anyhow::Result;
 use indicatif::{ProgressBar, ProgressStyle};
+use std::sync::{Arc, Mutex};
 use std::{fs, path::PathBuf, time::Duration};
 
 use globwalk::{GlobWalker, GlobWalkerBuilder};
@@ -29,60 +30,30 @@ impl Scanner {
     }
 
     pub fn build(app_args: &Params) -> Result<Self> {
-        let scan_directory = app_args.get_directory()?;
-        Ok(Scanner::new())
-            .map(|scanner| scanner.directory(scan_directory))
-            .map(|scanner| match app_args.get_min_size() {
-                Some(min_size) => scanner.min_size(min_size),
-                None => scanner,
-            })
-            .map(|scanner| match app_args.get_types() {
-                Some(ftypes) => scanner.filetypes(ftypes),
-                None => scanner,
-            })
-            .map(|scanner| match app_args.min_depth {
-                Some(min_depth) => scanner.min_depth(min_depth),
-                None => scanner,
-            })
-            .map(|scanner| match app_args.max_depth {
-                Some(max_depth) => scanner.max_depth(max_depth),
-                None => scanner,
-            })
+        let mut scanner = Scanner::new();
+        scanner.directory = Some(app_args.get_directory()?);
+
+        if let Some(min_size) = app_args.get_min_size() {
+            scanner.min_size = Some(min_size);
+        }
+
+        if app_args.get_types().is_some() {
+            scanner.filetypes = app_args.get_types();
+        }
+
+        if app_args.min_depth.is_some() {
+            scanner.min_depth = app_args.min_depth;
+        }
+
+        if app_args.max_depth.is_some() {
+            scanner.max_depth = app_args.max_depth;
+        }
+
+        Ok(scanner)
     }
 
-    pub fn min_size(&self, min_size: u64) -> Self {
-        Self {
-            min_size: Some(min_size),
-            ..self.clone()
-        }
-    }
-
-    pub fn min_depth(&self, min_depth: usize) -> Self {
-        Self {
-            min_depth: Some(min_depth),
-            ..self.clone()
-        }
-    }
-
-    pub fn max_depth(&self, max_depth: usize) -> Self {
-        Self {
-            max_depth: Some(max_depth),
-            ..self.clone()
-        }
-    }
-
-    pub fn directory(&self, dir: PathBuf) -> Self {
-        Self {
-            directory: Some(dir),
-            ..self.clone()
-        }
-    }
-
-    pub fn filetypes(&self, patterns: String) -> Self {
-        Self {
-            filetypes: Some(patterns),
-            ..self.clone()
-        }
+    pub fn filetypes(&mut self, patterns: String) {
+        self.filetypes = Some(patterns);
     }
 
     pub fn ignore_links(&self) -> Self {
@@ -144,7 +115,7 @@ impl Scanner {
         Ok(walker.build()?)
     }
 
-    pub fn scan(&self) -> Result<Vec<FileInfo>> {
+    pub fn scan(&self, files: Arc<Mutex<Vec<FileInfo>>>) -> Result<()> {
         let progress_style = ProgressStyle::with_template("[{elapsed_precise}] {pos:>7} {msg}")?;
         let progress_bar = ProgressBar::new_spinner();
         progress_bar.set_style(progress_style);
@@ -152,8 +123,7 @@ impl Scanner {
         progress_bar.set_message("paths mapped");
         let min_size = self.min_size.unwrap_or_default();
 
-        let results = self
-            .build_walker()?
+        self.build_walker()?
             .filter_map(Result::ok)
             .map(|entity| entity.into_path())
             .inspect(|_path| progress_bar.inc(1))
@@ -161,10 +131,12 @@ impl Scanner {
             .map(FileInfo::new)
             .filter_map(Result::ok)
             .filter(|file| file.size > min_size)
-            .collect::<Vec<FileInfo>>();
+            .for_each(|file| {
+                let mut flock = files.lock().unwrap();
+                flock.push(file);
+            });
 
         progress_bar.finish_with_message("paths mapped");
-
-        Ok(results)
+        Ok(())
     }
 }

src/server.rs

@@ -0,0 +1,7 @@
+pub struct Server {
+
+}
+
+impl Server {
+
+}