From d97af6b8f6835802f133a61acbf436bccc7c07d6 Mon Sep 17 00:00:00 2001 From: sreedevk Date: Mon, 14 Jul 2025 16:50:23 +0000 Subject: [PATCH] improved parallelization + updated benchmarks --- Cargo.toml | 2 +- README.md | 104 ++++++++++++++++++++++++++--------------------- src/fileinfo.rs | 21 +++++++++- src/formatter.rs | 6 ++- src/params.rs | 4 +- src/processor.rs | 84 ++++++++++++++++++++++---------------- src/server.rs | 24 +++++++---- 7 files changed, 152 insertions(+), 93 deletions(-) diff --git a/Cargo.toml b/Cargo.toml index b6b5453..e41501b 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -1,6 +1,6 @@ [package] name = "deduplicator" -version = "0.3.0" +version = "0.3.1" edition = "2021" description = "find,filter and delete duplicate files" repository = "https://github.com/sreedevk/deduplicator" diff --git a/README.md b/README.md index f8a6074..6e481b2 100644 --- a/README.md +++ b/README.md @@ -81,63 +81,71 @@ I've used hyperfine to run deduplicator on files generated by the rake file at ` ``` # hyperfine -N --warmup 80 './target/release/deduplicator bench_artifacts' Benchmark 1: ./target/release/deduplicator bench_artifacts - Time (mean ± σ): 2.5 ms ± 0.4 ms [User: 2.2 ms, System: 4.8 ms] - Range (min … max): 1.9 ms … 6.8 ms 1322 runs + Time (mean ± σ): 2.3 ms ± 0.5 ms [User: 2.4 ms, System: 4.6 ms] + Range (min … max): 1.4 ms … 6.7 ms 1457 runs -# dust 'bench_artifacts' - 37M ┌── file_1_fwds.bin │██ │ 1% -201M ├── file_0_fwds.bin │██████████ │ 8% -390M ├── file_0_fwdcbss.bin│████████████████████ │ 15% -390M ├── file_0_fwscas.bin │████████████████████ │ 15% -390M ├── file_0_fwss.bin │████████████████████ │ 15% -390M ├── file_1_fwdcbss.bin│████████████████████ │ 15% -390M ├── file_1_fwscas.bin │████████████████████ │ 15% -390M ├── file_1_fwss.bin │████████████████████ │ 15% -2.5G ┌─┴ bench_artifacts │███████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████ │ 100% +dust 'bench_artifacts' + 40M ┌── file_0_fwds.bin │████ │ 2% +113M ├── file_1_fwds.bin │███████████ │ 5% +390M ├── file_0_fwdcbss.bin│█████████████████████████████████████ │ 16% +390M ├── file_0_fwscas.bin │█████████████████████████████████████ │ 16% +390M ├── file_0_fwss.bin │█████████████████████████████████████ │ 16% +390M ├── file_1_fwdcbss.bin│█████████████████████████████████████ │ 16% +390M ├── file_1_fwscas.bin │█████████████████████████████████████ │ 16% +390M ├── file_1_fwss.bin │█████████████████████████████████████ │ 16% +2.4G ┌─┴ bench_artifacts │██████████████████████████████████████████████████████████████████ │ 100% ``` #### Many Small Files ``` # hyperfine --warmup 20 './target/release/deduplicator bench_artifacts' Benchmark 1: ./target/release/deduplicator bench_artifacts - Time (mean ± σ): 22.3 ms ± 1.7 ms [User: 35.8 ms, System: 46.3 ms] - Range (min … max): 18.9 ms … 27.2 ms 112 runs + Time (mean ± σ): 88.6 ms ± 10.3 ms [User: 275.3 ms, System: 232.7 ms] + Range (min … max): 66.8 ms … 108.0 ms 41 runs -# dust 'bench_artifacts' -3.9M ┌── file_992_fwss.bin │█ │ 0% -3.9M ├── file_993_fwdcbss.bin│█ │ 0% -3.9M ├── file_993_fwscas.bin │█ │ 0% -3.9M ├── file_993_fwss.bin │█ │ 0% -3.9M ├── file_994_fwdcbss.bin│█ │ 0% -3.9M ├── file_994_fwscas.bin │█ │ 0% -3.9M ├── file_994_fwss.bin │█ │ 0% -3.9M ├── file_995_fwdcbss.bin│█ │ 0% -3.9M ├── file_995_fwscas.bin │█ │ 0% -3.9M ├── file_995_fwss.bin │█ │ 0% -3.9M ├── file_996_fwdcbss.bin│█ │ 0% -3.9M ├── file_996_fwscas.bin │█ │ 0% -3.9M ├── file_996_fwss.bin │█ │ 0% -3.9M ├── file_997_fwdcbss.bin│█ │ 0% -3.9M ├── file_997_fwscas.bin │█ │ 0% -3.9M ├── file_997_fwss.bin │█ │ 0% -3.9M ├── file_998_fwdcbss.bin│█ │ 0% -3.9M ├── file_998_fwscas.bin │█ │ 0% -3.9M ├── file_998_fwss.bin │█ │ 0% -3.9M ├── file_999_fwdcbss.bin│█ │ 0% -3.9M ├── file_999_fwscas.bin │█ │ 0% -3.9M ├── file_999_fwss.bin │█ │ 0% -3.9M ├── file_99_fwdcbss.bin │█ │ 0% -3.9M ├── file_99_fwscas.bin │█ │ 0% -3.9M ├── file_99_fwss.bin │█ │ 0% -3.9M ├── file_9_fwdcbss.bin │█ │ 0% -3.9M ├── file_9_fwscas.bin │█ │ 0% -3.9M ├── file_9_fwss.bin │█ │ 0% - 11G ┌─┴ bench_artifacts │█████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████████ │ 100% +dust 'bench_artifacts' +3.9M ┌── file_98_fwss.bin │█ │ 0% +3.9M ├── file_990_fwdcbss.bin│█ │ 0% +3.9M ├── file_990_fwscas.bin │█ │ 0% +3.9M ├── file_990_fwss.bin │█ │ 0% +3.9M ├── file_991_fwdcbss.bin│█ │ 0% +3.9M ├── file_991_fwscas.bin │█ │ 0% +3.9M ├── file_991_fwss.bin │█ │ 0% +3.9M ├── file_992_fwdcbss.bin│█ │ 0% +3.9M ├── file_992_fwscas.bin │█ │ 0% +3.9M ├── file_992_fwss.bin │█ │ 0% +3.9M ├── file_993_fwdcbss.bin│█ │ 0% +3.9M ├── file_993_fwscas.bin │█ │ 0% +3.9M ├── file_993_fwss.bin │█ │ 0% +3.9M ├── file_994_fwdcbss.bin│█ │ 0% +3.9M ├── file_994_fwscas.bin │█ │ 0% +3.9M ├── file_994_fwss.bin │█ │ 0% +3.9M ├── file_995_fwdcbss.bin│█ │ 0% +3.9M ├── file_995_fwscas.bin │█ │ 0% +3.9M ├── file_995_fwss.bin │█ │ 0% +3.9M ├── file_996_fwdcbss.bin│█ │ 0% +3.9M ├── file_996_fwscas.bin │█ │ 0% +3.9M ├── file_996_fwss.bin │█ │ 0% +3.9M ├── file_997_fwdcbss.bin│█ │ 0% +3.9M ├── file_997_fwscas.bin │█ │ 0% +3.9M ├── file_997_fwss.bin │█ │ 0% +3.9M ├── file_998_fwdcbss.bin│█ │ 0% +3.9M ├── file_998_fwscas.bin │█ │ 0% +3.9M ├── file_998_fwss.bin │█ │ 0% +3.9M ├── file_999_fwdcbss.bin│█ │ 0% +3.9M ├── file_999_fwscas.bin │█ │ 0% +3.9M ├── file_999_fwss.bin │█ │ 0% +3.9M ├── file_99_fwdcbss.bin │█ │ 0% +3.9M ├── file_99_fwscas.bin │█ │ 0% +3.9M ├── file_99_fwss.bin │█ │ 0% +3.9M ├── file_9_fwdcbss.bin │█ │ 0% +3.9M ├── file_9_fwscas.bin │█ │ 0% +3.9M ├── file_9_fwss.bin │█ │ 0% + 11G ┌─┴ bench_artifacts │████████████████████████████████████████████████████████████████ │ 100% ``` ## proposed - [ ] parallelization - - [ ] (scanning + processing sw + processing hw) & formatting & printing - [ ] scanning + processing sw + processing hw + formatting + printing - [ ] max file path size should use the last set of duplicates - [ ] add more unit tests @@ -157,7 +165,11 @@ Benchmark 1: ./target/release/deduplicator bench_artifacts - [ ] --keep-last-modified - [ ] --keep-first-modified -## v0.3 +## v0.3.1 +- [x] parallelization + - [x] (scanning + processing sw + processing hw) & formatting & printing + +## v0.3.0 - [x] parallelization - [x] (scanning) + (processing sw & processing hw & formatting & printing) - [x] reduce cloning values on the heap diff --git a/src/fileinfo.rs b/src/fileinfo.rs index 1eb6a5b..d82fa9c 100644 --- a/src/fileinfo.rs +++ b/src/fileinfo.rs @@ -5,14 +5,22 @@ use std::{ fs, io::Read, path::{Path, PathBuf}, + sync::{Arc, Mutex}, time::SystemTime, }; +#[derive(Debug, Clone, PartialEq)] +pub enum FileState { + Unprocessed, + SwProcessed, +} + #[derive(Debug, Clone)] pub struct FileInfo { pub path: Box, pub size: u64, pub modified: SystemTime, + pub state: Arc>, } impl FileInfo { @@ -25,7 +33,7 @@ impl FileInfo { .fold(0u128, |acc, chunk: &[u8]| acc ^ gxhash128(chunk, seed))) } - pub fn initial_page_hash(&self, seed: i64) -> Result { + pub fn initpage_hash(&self, seed: i64) -> Result { let mut file = fs::File::open(&self.path)?; let mut buffer = [0; 4096]; let bytes_read = file.read(&mut buffer)?; @@ -39,6 +47,17 @@ impl FileInfo { path: path.into_boxed_path(), size: filemeta.len(), modified: filemeta.modified()?, + state: Arc::new(Mutex::new(FileState::Unprocessed)), }) } + + pub fn sw_processed(&self) { + let mut self_state = self.state.lock().unwrap(); + *self_state = FileState::SwProcessed; + } + + pub fn is_sw_processed(&self) -> bool { + let self_state = self.state.lock().unwrap(); + *self_state == FileState::SwProcessed + } } diff --git a/src/formatter.rs b/src/formatter.rs index 8da9ec0..989d624 100644 --- a/src/formatter.rs +++ b/src/formatter.rs @@ -36,6 +36,10 @@ impl Formatter { Ok(modified_time.format("%Y-%m-%d %H:%M:%S").to_string()) } + pub fn hash_hex(hash: &u128) -> Box { + format!("{:32x}", hash).into_boxed_str() + } + pub fn gen_sub_tbl(items: Vec, app_args: &Params, max_path_len: u64) -> Table { let mut inner_table = Table::new(); inner_table.set_format(*format::consts::FORMAT_NO_BORDER_LINE_SEPARATOR); @@ -71,7 +75,7 @@ impl Formatter { .filter(|i| i.value().len() > 1) .map(|i| { row![ - i.key(), + Self::hash_hex(i.key()), Self::gen_sub_tbl(i.value().to_vec(), args, mpath_len) ] }) diff --git a/src/params.rs b/src/params.rs index 675ac41..9d55316 100644 --- a/src/params.rs +++ b/src/params.rs @@ -96,6 +96,8 @@ mod tests { ..Default::default() }; - assert_eq!(params.get_types(), Some(String::from("pdf,tiff"))); + assert!(params + .get_types() + .is_some_and(|x| x == "pdf,tiff" || x == "tiff,pdf")) } } diff --git a/src/processor.rs b/src/processor.rs index b7ac840..88ac3f2 100644 --- a/src/processor.rs +++ b/src/processor.rs @@ -1,10 +1,8 @@ use anyhow::Result; use dashmap::DashMap; -use indicatif::{ - MultiProgress, ParallelProgressIterator, ProgressBar, ProgressFinish, ProgressStyle, -}; +use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; +use rayon::iter::IntoParallelRefMutIterator; use rayon::prelude::{IntoParallelIterator, ParallelIterator}; -use std::borrow::Cow; use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; use std::sync::{Arc, Mutex, TryLockError, TryLockResult}; use std::time::Duration; @@ -22,57 +20,68 @@ impl Processor { progress_bar_box: Arc, max_file_size: Arc, seed: i64, + sw_sorting_finished: Arc, ) -> Result<()> { let progress_bar = match app_args.progress { true => progress_bar_box.add(ProgressBar::new_spinner()), false => ProgressBar::hidden(), }; - let keys: Vec = sw_store.clone().iter().map(|i| *i.key()).collect(); - let progress_style = ProgressStyle::with_template("[{elapsed_precise}] {pos:>7} {msg}")?; progress_bar.set_style(progress_style); progress_bar.enable_steady_tick(Duration::from_millis(50)); progress_bar.set_message("files grouped by hash."); - keys.into_par_iter() - .progress_with(progress_bar) - .with_finish(ProgressFinish::WithMessage(Cow::from( - "files grouped by hash.", - ))) - .for_each(|key| { - let group: Vec = sw_store.get(&key).unwrap().to_vec(); - if group.len() > 1 { - group.into_par_iter().for_each(|file| { - let fhash = if app_args.strict { - file.hash(seed).expect("hashing file failed.") - } else { - file.initial_page_hash(seed).expect("hashing file failed.") - }; + loop { + let keys: Vec = sw_store + .clone() + .iter() + .filter(|i| !i.value().iter().all(|x| x.is_sw_processed())) + .filter(|i| i.value().len() > 1) + .map(|i| *i.key()) + .collect(); - Self::compare_and_update_max_path_len( - max_file_size.clone(), - file.path.to_string_lossy().len() as u64, - ) - .unwrap(); - - hw_store - .entry(fhash) - .and_modify(|fileset| fileset.push(file.clone())) - .or_insert_with(|| vec![file]); - }); + if keys.is_empty() { + match sw_sorting_finished.load(std::sync::atomic::Ordering::Relaxed) { + true => { + progress_bar.finish_with_message("files grouped by hash."); + break Ok(()); + } + false => continue, } - }); + } else { + keys.into_par_iter().for_each(|key| { + let mut group: Vec = sw_store.get(&key).unwrap().to_vec(); + if group.len() > 1 { + group.par_iter_mut().for_each(|file| { + progress_bar.inc(1); + file.sw_processed(); - Ok(()) + let fhash = match app_args.strict { + true => file.hash(seed).expect("hashing file failed."), + false => file.initpage_hash(seed).expect("hashing file failed."), + }; + + Self::compare_and_update_max_path_len( + max_file_size.clone(), + file.path.to_string_lossy().len() as u64, + ); + + hw_store + .entry(fhash) + .and_modify(|fileset| fileset.push(file.clone())) + .or_insert_with(|| vec![file.clone()]); + }); + }; + }); + } + } } - pub fn compare_and_update_max_path_len(current: Arc, next: u64) -> Result<()> { + pub fn compare_and_update_max_path_len(current: Arc, next: u64) { if current.load(Ordering::Relaxed) < next { current.store(next, Ordering::Release); } - - Ok(()) } pub fn sizewise( @@ -193,6 +202,7 @@ mod tests { Arc::new(MultiProgress::new()), Arc::new(AtomicU64::new(32)), 300, + Arc::new(AtomicBool::new(true)), )?; assert_eq!(hw_dupstore.len(), 2); @@ -245,6 +255,7 @@ mod tests { Arc::new(MultiProgress::new()), Arc::new(AtomicU64::new(32)), 300, + Arc::new(AtomicBool::new(true)), )?; assert_eq!(hw_dupstore.len(), 1); @@ -290,6 +301,7 @@ mod tests { Arc::new(MultiProgress::new()), Arc::new(AtomicU64::new(32)), 300, + Arc::new(AtomicBool::new(true)), )?; assert_eq!(hw_dupstore.len(), 1); diff --git a/src/server.rs b/src/server.rs index 6dcb1ac..1d02f3e 100644 --- a/src/server.rs +++ b/src/server.rs @@ -43,20 +43,27 @@ impl Server { } let app_args_clone_for_sc = self.app_args.clone(); - let app_args_clone_for_pr = self.app_args.clone(); + let app_args_clone_for_sw = self.app_args.clone(); + let app_args_clone_for_hw = 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 sw_sort_finished = Arc::new(AtomicBool::new(false)); let sfin_sc_tr_cl = scanner_finished.clone(); let sfin_pr_tr_cl = scanner_finished.clone(); + let swfin_pr_tr_sw = sw_sort_finished.clone(); + let swfin_pr_tr_hw = sw_sort_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(); let progbarbox_sc_clone = progbarbox.clone(); + let progbarbox_pr_clone_for_sw = progbarbox.clone(); + let progbarbox_pr_clone_for_hw = progbarbox.clone(); self.threadpool.execute(move || { Scanner::new(app_args_clone_for_sc) @@ -67,25 +74,28 @@ impl Server { sfin_sc_tr_cl.store(true, std::sync::atomic::Ordering::Relaxed); }); - let progbarbox_pr_clone = progbarbox.clone(); - self.threadpool.execute(move || { Processor::sizewise( - app_args_clone_for_pr.clone(), + app_args_clone_for_sw, sfin_pr_tr_cl, store_dupl_sw_for_sw, file_queue_clone_pr, - progbarbox_pr_clone.clone(), + progbarbox_pr_clone_for_sw, ) .unwrap(); + swfin_pr_tr_sw.store(true, std::sync::atomic::Ordering::Relaxed); + }); + + self.threadpool.execute(move || { Processor::hashwise( - app_args_clone_for_pr, + app_args_clone_for_hw, store_dupl_sw_for_hw, store_dupl_hw, - progbarbox_pr_clone, + progbarbox_pr_clone_for_hw, max_file_path_len_clone, seed, + swfin_pr_tr_hw, ) .unwrap(); });