parallelization improvements

Cargo.lock

@@ -291,6 +291,7 @@ dependencies = [
  "rayon",
  "serde",
  "serde_json",
+ "threadpool",
  "unicode-segmentation",
 ]
 
@@ -395,6 +396,12 @@ version = "0.3.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "d231dfb89cfffdbc30e7fc41579ed6066ad03abda9e567ccafae602b97ec5024"
 
+[[package]]
+name = "hermit-abi"
+version = "0.5.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c"
+
 [[package]]
 name = "iana-time-zone"
 version = "0.1.60"
@@ -463,7 +470,7 @@ version = "0.4.12"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "f23ff5ef2b80d608d61efee834934d862cd92461afc0560dedf493e4c033738b"
 dependencies = [
- "hermit-abi",
+ "hermit-abi 0.3.9",
  "libc",
  "windows-sys",
 ]
@@ -560,6 +567,16 @@ dependencies = [
  "autocfg",
 ]
 
+[[package]]
+name = "num_cpus"
+version = "1.17.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b"
+dependencies = [
+ "hermit-abi 0.5.2",
+ "libc",
+]
+
 [[package]]
 name = "number_prefix"
 version = "0.4.0"
@@ -798,6 +815,15 @@ dependencies = [
  "syn",
 ]
 
+[[package]]
+name = "threadpool"
+version = "1.8.1"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d050e60b33d41c19108b32cea32164033a9013fe3b46cbd4457559bfbf77afaa"
+dependencies = [
+ "num_cpus",
+]
+
 [[package]]
 name = "unicode-ident"
 version = "1.0.12"

Cargo.toml

@@ -29,6 +29,7 @@ prettytable-rs = "0.10.0"
 rayon = "1.6.1"
 serde = { version = "1.0.192", features = ["derive"] }
 serde_json = "1.0.108"
+threadpool = "1.8.1"
 unicode-segmentation = "1.10.0"
 
 [profile.release]

src/formatter.rs

@@ -9,6 +9,7 @@ use pathdiff::diff_paths;
 use prettytable::{format, row, Table};
 use std::borrow::Cow;
 use std::path::PathBuf;
+use std::sync::Arc;
 use std::time::Duration;
 
 impl Formatter {
@@ -38,7 +39,11 @@ impl Formatter {
         Ok(modified_time.format("%Y-%m-%d %H:%M:%S").to_string())
     }
 
-    pub fn generate_table(raw: DashMap<String, Vec<FileInfo>>, max_path_len: usize, app_args: &Params) -> Result<Table> {
+    pub fn generate_table(
+        raw: Arc<DashMap<String, Vec<FileInfo>>>,
+        max_path_len: u64,
+        app_args: &Params,
+    ) -> Result<Table> {
         let mut output_table = Table::new();
         output_table.set_titles(row!["hash", "duplicates"]);
 
@@ -51,27 +56,31 @@ impl Formatter {
         progress_bar.enable_steady_tick(Duration::from_millis(50));
         progress_bar.set_message("generating output");
 
-        raw.into_iter()
+        raw.iter()
             .progress_with(progress_bar)
             .with_finish(ProgressFinish::WithMessage(Cow::from("output generated")))
-            .for_each(|(hash, group)| {
+            .for_each(|i| {
                 let mut inner_table = Table::new();
                 inner_table.set_format(*format::consts::FORMAT_NO_BORDER_LINE_SEPARATOR);
-                group.iter().for_each(|file| {
+                i.value().iter().for_each(|file| {
                     inner_table.add_row(row![
-                        Self::human_path(file, app_args, max_path_len).unwrap_or_default(),
+                        Self::human_path(file, app_args, max_path_len as usize).unwrap_or_default(),
                         Self::human_filesize(file).unwrap_or_default(),
                         Self::human_mtime(file).unwrap_or_default()
                     ]);
                 });
 
-                output_table.add_row(row![hash, inner_table]);
+                output_table.add_row(row![i.key(), inner_table]);
             });
 
         Ok(output_table)
     }
 
-    pub fn print(raw: DashMap<String, Vec<FileInfo>>, max_path_len: usize, app_args: &Params) -> Result<()> {
+    pub fn print(
+        raw: Arc<DashMap<String, Vec<FileInfo>>>,
+        max_path_len: u64,
+        app_args: &Params,
+    ) -> Result<()> {
         if raw.is_empty() {
             println!("\n\nNo duplicates found matching your search criteria.\n");
             return Ok(());

src/interactive.rs

@@ -4,6 +4,7 @@ use anyhow::Result;
 use dashmap::DashMap;
 use prettytable::{format, row, Table};
 use std::io::{self, Write};
+use std::sync::Arc;
 
 pub fn scan_group_confirmation() -> Result<bool> {
     print!("\nconfirm? [y/N]: ");
@@ -29,33 +30,30 @@ pub fn scan_group_instruction() -> Result<String> {
     Ok(user_input)
 }
 
-pub fn init(result: DashMap<String, Vec<FileInfo>>, app_args: &Params) -> Result<()> {
-    result
-        .clone()
-        .into_iter()
-        .enumerate()
-        .for_each(|(gindex, (_, group))| {
-            let mut itable = Table::new();
-            itable.set_format(*format::consts::FORMAT_NO_BORDER_LINE_SEPARATOR);
-            itable.set_titles(row!["index", "filename", "size", "updated_at"]);
-            let max_path_size = group
-                .iter()
-                .map(|f| f.path.clone().into_os_string().len())
-                .max()
-                .unwrap_or_default();
+pub fn init(result: Arc<DashMap<String, Vec<FileInfo>>>, app_args: &Params) -> Result<()> {
+    result.clone().iter().enumerate().for_each(|(gindex, i)| {
+        let group = i.value();
+        let mut itable = Table::new();
+        itable.set_format(*format::consts::FORMAT_NO_BORDER_LINE_SEPARATOR);
+        itable.set_titles(row!["index", "filename", "size", "updated_at"]);
+        let max_path_size = group
+            .iter()
+            .map(|f| f.path.clone().into_os_string().len())
+            .max()
+            .unwrap_or_default();
 
-            group.iter().enumerate().for_each(|(index, file)| {
-                itable.add_row(row![
-                    index,
-                    Formatter::human_path(file, app_args, max_path_size).unwrap_or_default(),
-                    Formatter::human_filesize(file).unwrap_or_default(),
-                    Formatter::human_mtime(file).unwrap_or_default()
-                ]);
-            });
-
-            process_group_action(&group, gindex, result.len(), itable);
+        group.iter().enumerate().for_each(|(index, file)| {
+            itable.add_row(row![
+                index,
+                Formatter::human_path(file, app_args, max_path_size).unwrap_or_default(),
+                Formatter::human_filesize(file).unwrap_or_default(),
+                Formatter::human_mtime(file).unwrap_or_default()
+            ]);
         });
 
+        process_group_action(&group, gindex, result.len(), itable);
+    });
+
     Ok(())
 }
 

src/main.rs

@@ -6,50 +6,32 @@ mod processor;
 mod scanner;
 mod server;
 
+use std::sync::atomic::Ordering;
+
+use self::formatter::Formatter;
+use self::server::Server;
 use anyhow::Result;
 use clap::Parser;
-use formatter::Formatter;
 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 server = Server::new(app_args.clone());
 
-    // TODO: RACE CONDITION BETWEEN SCANNER AND PROCESSOR
-    let file_arc_clone = files.clone();
-    let app_args_clone = app_args.clone();
-
-    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();
+    server.start()?;
 
     match app_args.interactive {
-        false => Formatter::print(results.0, results.1, &app_args)?,
-        true => interactive::init(results.0, &app_args)?,
-    }
+        false => {
+            Formatter::print(
+                server.hw_duplicate_set,
+                server.max_file_path_len.load(Ordering::Acquire),
+                &app_args,
+            )?;
+        }
+        true => {
+            interactive::init(server.hw_duplicate_set, &app_args)?;
+        }
+    };
 
     Ok(())
 }

src/processor.rs

@@ -1,118 +1,72 @@
 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 std::sync::atomic::{AtomicBool, AtomicU64, Ordering};
+use std::sync::{Arc, Mutex, TryLockError, TryLockResult};
 
 use crate::fileinfo::FileInfo;
 
-#[derive(Debug, Clone)]
-pub struct Processor {
-    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,
-}
+pub struct Processor {}
 
 impl Processor {
-    pub fn new(files: Arc<Mutex<Vec<FileInfo>>>) -> Self {
-        Self {
-            files,
-            hashwise_results: DashMap::new(),
-            sizewise_results: DashMap::new(),
-            max_path_len: 0,
-        }
-    }
+    pub fn hashwise(
+        sw_store: Arc<DashMap<u64, Vec<FileInfo>>>,
+        hw_store: Arc<DashMap<String, Vec<FileInfo>>>,
+    ) -> Result<()> {
+        let keys: Vec<u64> = sw_store.clone().iter().map(|i| *i.key()).collect();
 
-    pub fn hashwise(&mut self) -> Result<()> {
-        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(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");
-
-        let filelist = self
-            .sizewise_results
-            .clone()
-            .into_read_only()
-            .values()
-            .filter(|&subfiles| subfiles.len() > 1)
-            .flatten()
-            .cloned()
-            .collect::<Vec<FileInfo>>();
-
-        self.max_path_len = filelist
-            .iter()
-            .map(|x| x.path.clone().into_os_string().len())
-            .max()
-            .unwrap_or_default();
-
-        filelist
-            .into_par_iter()
-            .progress_with(progress_bar)
-            .with_finish(ProgressFinish::WithMessage(Cow::from(
-                "indexed files hashes",
-            )))
-            .map(|file| file.hash())
-            .filter_map(Result::ok)
-            .for_each(move |file| {
-                self.hashwise_results
-                    .entry(file.hash.clone().unwrap_or_default())
-                    .and_modify(|fileset| fileset.push(file.clone()))
-                    .or_insert_with(|| vec![file]);
-            });
+        keys.into_iter().for_each(|key| {
+            let group: Vec<FileInfo> = sw_store.get(&key).unwrap().to_vec();
+            if group.len() > 1 {
+                group.into_par_iter().for_each(|file| {
+                    hw_store
+                        .entry(file.hash.clone().unwrap_or_default())
+                        .and_modify(|fileset| fileset.push(file.clone()))
+                        .or_insert_with(|| vec![file]);
+                });
+            }
+        });
 
         Ok(())
     }
 
-    pub fn sizewise(&mut self) -> Result<()> {
-        let flist_size = {
-            let f = self.files.lock().unwrap();
-            f.len()
-        };
-
-        if flist_size < 1 {
-            return Ok(());
+    pub fn compare_and_update_max_path_len(current: Arc<AtomicU64>, next: u64) -> Result<()> {
+        if current.load(Ordering::Relaxed) < next {
+            current.store(next, Ordering::Release);
         }
 
-        let progress_style = ProgressStyle::with_template(
-            "[{elapsed_precise}] {bar:40.cyan/blue} {pos:>7}/{len:7} {msg}",
-        )?;
-        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");
-
-        (0..flist_size)
-            .into_par_iter()
-            .progress_with(progress_bar)
-            .with_finish(ProgressFinish::WithMessage(Cow::from(
-                "indexed files sizes",
-            )))
-            .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()))
-                    .or_insert_with(|| vec![file]);
-            });
-
         Ok(())
     }
+
+    // TODO: reduce the amount of time files remain locked for
+    pub fn sizewise(
+        scanner_finished: Arc<AtomicBool>,
+        store: Arc<DashMap<u64, Vec<FileInfo>>>,
+        files: Arc<Mutex<Vec<FileInfo>>>,
+        max_file_size: Arc<AtomicU64>,
+    ) -> Result<()> {
+        loop {
+            match files.try_lock() {
+                Ok(mut flist) => match flist.pop() {
+                    Some(file) => {
+                        Self::compare_and_update_max_path_len(
+                            max_file_size.clone(),
+                            file.path.to_string_lossy().len() as u64,
+                        )?;
+                        store
+                            .entry(file.size)
+                            .and_modify(|fileset| fileset.push(file.clone()))
+                            .or_insert_with(|| vec![file]);
+                        continue;
+                    }
+                    None => match scanner_finished.load(std::sync::atomic::Ordering::Relaxed) {
+                        true => break Ok(()),
+                        false => continue,
+                    },
+                },
+                TryLockResult::Err(TryLockError::WouldBlock) => continue,
+                _ => break Ok(()),
+            }
+        }
+    }
 }

src/scanner.rs

@@ -29,7 +29,7 @@ impl Scanner {
         }
     }
 
-    pub fn build(app_args: &Params) -> Result<Self> {
+    pub fn build(app_args: Arc<Params>) -> Result<Self> {
         let mut scanner = Scanner::new();
         scanner.directory = Some(app_args.get_directory()?);
 

src/server.rs

@@ -1,7 +1,73 @@
-pub struct Server {
+use std::sync::atomic::{AtomicBool, AtomicU64};
+use std::sync::{Arc, Mutex};
 
+use crate::processor::Processor;
+use crate::scanner::Scanner;
+use anyhow::Result;
+use dashmap::DashMap;
+use threadpool::ThreadPool;
+
+use crate::fileinfo::FileInfo;
+use crate::params::Params;
+
+pub struct Server {
+    filequeue: Arc<Mutex<Vec<FileInfo>>>,
+    sw_duplicate_set: Arc<DashMap<u64, Vec<FileInfo>>>,
+    pub hw_duplicate_set: Arc<DashMap<String, Vec<FileInfo>>>,
+    threadpool: ThreadPool,
+    app_args: Arc<Params>,
+    pub max_file_path_len: Arc<AtomicU64>,
 }
 
 impl Server {
+    pub fn new(opts: Params) -> Self {
+        Self {
+            filequeue: Arc::new(Mutex::new(Vec::new())),
+            sw_duplicate_set: Arc::new(DashMap::new()),
+            hw_duplicate_set: Arc::new(DashMap::new()),
+            threadpool: ThreadPool::new(8),
+            app_args: Arc::new(opts),
+            max_file_path_len: Arc::new(AtomicU64::new(0)),
+        }
+    }
 
+    pub fn start(&self) -> Result<()> {
+        let app_args_clone = self.app_args.clone();
+        let file_queue_clone_sc = self.filequeue.clone();
+        let file_queue_clone_pr = self.filequeue.clone();
+        let scanner_finished = Arc::new(AtomicBool::new(false));
+
+        let sfin_sc_tr_cl = scanner_finished.clone();
+        let sfin_pr_tr_cl = scanner_finished.clone();
+
+        let store_dupl_sw_for_sw = self.sw_duplicate_set.clone();
+        let store_dupl_sw_for_hw = self.sw_duplicate_set.clone();
+        let store_dupl_hw = self.hw_duplicate_set.clone();
+        let max_file_path_len_clone = self.max_file_path_len.clone();
+
+        self.threadpool.execute(move || {
+            Scanner::build(app_args_clone)
+                .unwrap()
+                .scan(file_queue_clone_sc)
+                .unwrap();
+
+            sfin_sc_tr_cl.store(true, std::sync::atomic::Ordering::Relaxed);
+        });
+
+        self.threadpool.execute(move || {
+            Processor::sizewise(
+                sfin_pr_tr_cl,
+                store_dupl_sw_for_sw,
+                file_queue_clone_pr,
+                max_file_path_len_clone,
+            )
+            .unwrap();
+
+            Processor::hashwise(store_dupl_sw_for_hw, store_dupl_hw).unwrap();
+        });
+
+        self.threadpool.join();
+
+        Ok(())
+    }
 }