From 4a6aadb78e29d76b7a9933ebc0824e98dd4fd6c6 Mon Sep 17 00:00:00 2001 From: sreedevk Date: Sat, 12 Jul 2025 20:21:44 +0000 Subject: [PATCH] parallelization improvements --- Cargo.lock | 28 ++++++++- Cargo.toml | 1 + src/formatter.rs | 23 ++++--- src/interactive.rs | 46 +++++++------- src/main.rs | 52 +++++---------- src/processor.rs | 154 ++++++++++++++++----------------------------- src/scanner.rs | 2 +- src/server.rs | 68 +++++++++++++++++++- 8 files changed, 205 insertions(+), 169 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 2551cc9..70697e9 100644 --- a/Cargo.lock +++ b/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" diff --git a/Cargo.toml b/Cargo.toml index 17b0944..6f49017 100644 --- a/Cargo.toml +++ b/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] diff --git a/src/formatter.rs b/src/formatter.rs index a34ce11..ed2b2a6 100644 --- a/src/formatter.rs +++ b/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>, max_path_len: usize, app_args: &Params) -> Result { + pub fn generate_table( + raw: Arc>>, + max_path_len: u64, + app_args: &Params, + ) -> Result
{ 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>, max_path_len: usize, app_args: &Params) -> Result<()> { + pub fn print( + raw: Arc>>, + max_path_len: u64, + app_args: &Params, + ) -> Result<()> { if raw.is_empty() { println!("\n\nNo duplicates found matching your search criteria.\n"); return Ok(()); diff --git a/src/interactive.rs b/src/interactive.rs index 5b36282..f6cebce 100644 --- a/src/interactive.rs +++ b/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 { print!("\nconfirm? [y/N]: "); @@ -29,33 +30,30 @@ pub fn scan_group_instruction() -> Result { Ok(user_input) } -pub fn init(result: DashMap>, 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>>, 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(()) } diff --git a/src/main.rs b/src/main.rs index dd7540a..cf9402f 100644 --- a/src/main.rs +++ b/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>> = 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(()) } diff --git a/src/processor.rs b/src/processor.rs index 0e0a9f3..78e27b5 100644 --- a/src/processor.rs +++ b/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>>, - pub hashwise_results: DashMap>, - pub sizewise_results: DashMap>, - pub max_path_len: usize, -} +pub struct Processor {} impl Processor { - pub fn new(files: Arc>>) -> Self { - Self { - files, - hashwise_results: DashMap::new(), - sizewise_results: DashMap::new(), - max_path_len: 0, - } - } + pub fn hashwise( + sw_store: Arc>>, + hw_store: Arc>>, + ) -> Result<()> { + let keys: Vec = 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::>(); - - 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 = 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, 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, + store: Arc>>, + files: Arc>>, + max_file_size: Arc, + ) -> 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(()), + } + } + } } diff --git a/src/scanner.rs b/src/scanner.rs index 11d0cc0..ae0f834 100644 --- a/src/scanner.rs +++ b/src/scanner.rs @@ -29,7 +29,7 @@ impl Scanner { } } - pub fn build(app_args: &Params) -> Result { + pub fn build(app_args: Arc) -> Result { let mut scanner = Scanner::new(); scanner.directory = Some(app_args.get_directory()?); diff --git a/src/server.rs b/src/server.rs index e20fb20..0c2ed96 100644 --- a/src/server.rs +++ b/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>>, + sw_duplicate_set: Arc>>, + pub hw_duplicate_set: Arc>>, + threadpool: ThreadPool, + app_args: Arc, + pub max_file_path_len: Arc, } 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(()) + } }