From 3ca931fa8e70dd9c201baa0c16f988b97ec09a1c Mon Sep 17 00:00:00 2001 From: Sreedev Kodichath Date: Sun, 26 Jul 2026 14:27:02 +0200 Subject: [PATCH] [REF] core: replace streaming pipeline with staged batch pipeline The deduplication core was a three-stage streaming producer/consumer (scan -> group-by-size -> group-by-hash) orchestrated by a Server struct that ran the stages concurrently on a threadpool. Coordination relied on AtomicBool flags polled in busy-wait loops, an Arc> hand-off queue, DashMap stores, and a per-file Arc>. Every stage had to receive its own hand-cloned Arcs with ad-hoc names, and the busy-wait branches burned a core spinning while waiting for the producer. Replace it with a staged batch pipeline over owned collections: pipeline::run(&Params) drives scan -> group_by_size -> group_by_hash in sequence, with rayon supplying parallelism. With no shared mutable state between stages, all the Arc plumbing, the AtomicBool coordination, the Mutex queue, the per-file lock, server.rs, and the threadpool/dashmap dependencies are gone. FileInfo becomes plain data and each stage is a pure, independently testable function. Behavior is preserved except for three authorized deviations: progress spinners render sequentially rather than concurrently under -p; the interactive empty-result message prints once instead of twice; and the interactive "Duplicate Set X of Y" total now counts the confirmed duplicate groups shown instead of the internal candidate-hash store size. This lands as the foundation for the cache, CLI, and TUI work that follows. --- .gitignore | 3 + Cargo.lock | 86 --------- Cargo.toml | 2 - src/fileinfo.rs | 19 -- src/formatter.rs | 76 ++++---- src/interactive.rs | 64 +++---- src/main.rs | 27 +-- src/pipeline.rs | 97 ++++++++++ src/processor.rs | 453 ++++++++++++--------------------------------- src/scanner.rs | 104 +++-------- src/server.rs | 117 ------------ 11 files changed, 316 insertions(+), 732 deletions(-) create mode 100644 src/pipeline.rs delete mode 100644 src/server.rs diff --git a/.gitignore b/.gitignore index c89a500..98c71c9 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,6 @@ /Cargo.lock /result-bin /.bacon-locations +/docs +/.claude +/.superpowers diff --git a/Cargo.lock b/Cargo.lock index 99a277c..d14ea8a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -256,21 +256,6 @@ dependencies = [ "memchr", ] -[[package]] -name = "dashmap" -version = "6.1.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5041cc499144891f3790297212f32a74fb938e5136a14943f338ef9e0ae276cf" -dependencies = [ - "cfg-if", - "crossbeam-utils", - "hashbrown", - "lock_api", - "once_cell", - "parking_lot_core", - "rayon", -] - [[package]] name = "deduplicator" version = "0.3.2" @@ -279,7 +264,6 @@ dependencies = [ "bytesize", "chrono", "clap", - "dashmap", "globwalk", "gxhash", "indicatif", @@ -289,7 +273,6 @@ dependencies = [ "rand", "rayon", "tempfile", - "threadpool", "unicode-segmentation", ] @@ -398,12 +381,6 @@ dependencies = [ "rustversion", ] -[[package]] -name = "hashbrown" -version = "0.14.5" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e5274423e17b7c9fc20b6e7e208532f9b19825d82dfd615708b70edd83df41f1" - [[package]] name = "heck" version = "0.5.0" @@ -531,16 +508,6 @@ version = "0.9.4" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "cd945864f07fe9f5371a27ad7b52a172b4b499999f1d97574c9fa68373937e12" -[[package]] -name = "lock_api" -version = "0.4.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "96936507f153605bddfcda068dd804796c84324ed2510809e5b2a624c81da765" -dependencies = [ - "autocfg", - "scopeguard", -] - [[package]] name = "log" version = "0.4.27" @@ -571,16 +538,6 @@ 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", - "libc", -] - [[package]] name = "once_cell" version = "1.21.3" @@ -593,19 +550,6 @@ version = "1.70.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a4895175b425cb1f87721b59f0f286c2092bd4af812243672510e1ac53e2e0ad" -[[package]] -name = "parking_lot_core" -version = "0.9.11" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "bc838d2a56b5b1a6c25f55575dfc605fabb63bb2365f6c2353ef9159aa69e4a5" -dependencies = [ - "cfg-if", - "libc", - "redox_syscall", - "smallvec", - "windows-targets 0.52.6", -] - [[package]] name = "pathdiff" version = "0.2.3" @@ -714,15 +658,6 @@ dependencies = [ "crossbeam-utils", ] -[[package]] -name = "redox_syscall" -version = "0.5.13" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "0d04b7d0ee6b4a0207a0a7adb104d23ecb0b47d6beae7152d0fa34b692b29fd6" -dependencies = [ - "bitflags", -] - [[package]] name = "redox_users" version = "0.4.6" @@ -785,12 +720,6 @@ dependencies = [ "winapi-util", ] -[[package]] -name = "scopeguard" -version = "1.2.0" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "94143f37725109f92c262ed2cf5e59bce7498c01bcc1502d7b9afe439a4e9f49" - [[package]] name = "serde" version = "1.0.219" @@ -817,12 +746,6 @@ version = "1.3.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "0fda2ff0d084019ba4d7c6f371c95d8fd75ce3524c3cb8fb653a3023f6323e64" -[[package]] -name = "smallvec" -version = "1.15.1" -source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "67b1b7a3b5fe4f1376887184045fcf45c69e92af734b7aaddc05fb777b6fbd03" - [[package]] name = "strsim" version = "0.11.1" @@ -884,15 +807,6 @@ 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.18" diff --git a/Cargo.toml b/Cargo.toml index 30ebf2f..7314661 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -21,7 +21,6 @@ anyhow = "1.0.68" bytesize = "2.0.1" chrono = "0.4.23" clap = { version = "4.0.32", features = ["derive"] } -dashmap = { version = "6.1.0", features = ["rayon"] } globwalk = "0.9.1" gxhash = { version = "3.4.1", default-features = false } indicatif = { version = "0.18.0", features = ["rayon"] } @@ -30,7 +29,6 @@ pathdiff = "0.2.1" prettytable-rs = "0.10.0" rand = "0.9.1" rayon = "1.6.1" -threadpool = "1.8.1" unicode-segmentation = "1.12.0" [profile.release] diff --git a/src/fileinfo.rs b/src/fileinfo.rs index f0d091c..4721660 100644 --- a/src/fileinfo.rs +++ b/src/fileinfo.rs @@ -5,22 +5,14 @@ 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 { @@ -53,19 +45,8 @@ 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 - } } #[cfg(test)] diff --git a/src/formatter.rs b/src/formatter.rs index 29d61b1..12143d7 100644 --- a/src/formatter.rs +++ b/src/formatter.rs @@ -1,11 +1,9 @@ -use crate::{fileinfo::FileInfo, params::Params}; +use crate::{fileinfo::FileInfo, params::Params, pipeline::DedupReport}; use anyhow::Result; use chrono::{DateTime, Utc}; -use dashmap::DashMap; use pathdiff::diff_paths; use rayon::prelude::*; -use std::sync::atomic::AtomicU64; -use std::{path::PathBuf, sync::Arc}; +use std::path::PathBuf; const YELLOW: &str = "\x1b[33m"; const RESET: &str = "\x1b[0m"; @@ -34,47 +32,39 @@ impl Formatter { Ok(modified_time.format("%Y-%m-%d %H:%M:%S").to_string()) } - pub fn print(raw: Arc>>, max_path_len: u64, aargs: &Params) { - print!("{}", "\n".repeat(if aargs.progress { 2 } else { 1 })); // spacing + pub fn print(report: &DedupReport, aargs: &Params) { + print!("{}", "\n".repeat(if aargs.progress { 2 } else { 1 })); - if raw.is_empty() { + if report.groups.is_empty() { println!("No duplicates found matching your search criteria."); - } else { - let printed_count: AtomicU64 = AtomicU64::new(0); - - raw.par_iter().for_each(|sref| { - if sref.value().len() > 1 { - printed_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - let mut ostring = format!("{}{:32x}{}\n", YELLOW, sref.key(), RESET); - let subfields = sref - .value() - .par_iter() - .enumerate() - .map(|(i, finfo)| { - let nodechar = if i == sref.value().len() - 1 { - "└─" - } else { - "├─" - }; - format!( - "{}\t{}\t{}\t{}\n", - nodechar, - Self::human_path(finfo, aargs, max_path_len as usize) - .expect("path formatting failed."), - Self::human_filesize(finfo).expect("filesize formatting failed."), - Self::human_mtime(finfo).expect("modified time formatting failed.") - ) - }) - .collect::(); - - ostring.push_str(&subfields); - println!("{ostring}"); - } - }); - - if printed_count.load(std::sync::atomic::Ordering::Relaxed) < 1 { - println!("No duplicates found matching your search criteria."); - } + return; } + + report.groups.par_iter().for_each(|group| { + let mut ostring = format!("{}{:32x}{}\n", YELLOW, group.hash, RESET); + let subfields = group + .files + .par_iter() + .enumerate() + .map(|(i, finfo)| { + let nodechar = if i == group.files.len() - 1 { + "└─" + } else { + "├─" + }; + format!( + "{}\t{}\t{}\t{}\n", + nodechar, + Self::human_path(finfo, aargs, report.max_path_len) + .expect("path formatting failed."), + Self::human_filesize(finfo).expect("filesize formatting failed."), + Self::human_mtime(finfo).expect("modified time formatting failed.") + ) + }) + .collect::(); + + ostring.push_str(&subfields); + println!("{ostring}"); + }); } } diff --git a/src/interactive.rs b/src/interactive.rs index 3cb20cd..2600077 100644 --- a/src/interactive.rs +++ b/src/interactive.rs @@ -1,56 +1,40 @@ -use crate::{fileinfo::FileInfo, formatter::Formatter, params::Params}; +use crate::{fileinfo::FileInfo, formatter::Formatter, params::Params, pipeline::DuplicateGroup}; use anyhow::Result; -use dashmap::DashMap; use prettytable::{format, row, Table}; -use std::sync::atomic::AtomicU64; -use std::{ - io::{self, Write}, - sync::Arc, -}; +use std::io::{self, Write}; pub struct Interactive; impl Interactive { - pub fn init(result: Arc>>, app_args: &Params) -> Result<()> { - let store = result.clone(); - if store.is_empty() { + pub fn init(groups: &[DuplicateGroup], app_args: &Params) -> Result<()> { + if groups.is_empty() { println!("No duplicates found matching your search criteria."); + return Ok(()); } - let printed_count: AtomicU64 = AtomicU64::new(0); + groups.iter().enumerate().for_each(|(gindex, group)| { + let files = &group.files; + let mut itable = Table::new(); + itable.set_format(*format::consts::FORMAT_NO_BORDER_LINE_SEPARATOR); + itable.set_titles(row!["index", "filename", "size", "updated_at"]); - store - .iter() - .filter(|i| i.value().len() > 1) - .enumerate() - .for_each(|(gindex, i)| { - printed_count.fetch_add(1, std::sync::atomic::Ordering::Relaxed); - 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 = files + .iter() + .map(|f| f.path.iter().count()) + .max() + .unwrap_or_default(); - let max_path_size = group - .iter() - .map(|f| f.path.iter().count()) - .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() - ]); - }); - - Self::process_group_action(group, gindex, result.len(), itable); + files.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() + ]); }); - if printed_count.load(std::sync::atomic::Ordering::Relaxed) < 1 { - println!("No duplicates found matching your search criteria."); - } + Self::process_group_action(files, gindex, groups.len(), itable); + }); Ok(()) } diff --git a/src/main.rs b/src/main.rs index c498388..90595d1 100644 --- a/src/main.rs +++ b/src/main.rs @@ -2,34 +2,23 @@ mod fileinfo; mod formatter; mod interactive; mod params; +mod pipeline; mod processor; mod scanner; -mod server; -use self::{formatter::Formatter, interactive::Interactive, server::Server}; +use self::{formatter::Formatter, interactive::Interactive}; use anyhow::Result; use clap::Parser; use params::Params; -use std::sync::atomic::Ordering; fn main() -> Result<()> { - let app_args = Params::parse(); - let server = Server::new(app_args.clone()); + let params = Params::parse(); + let report = pipeline::run(¶ms)?; - server.start()?; - - match app_args.interactive { - 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)?; - } - }; + match params.interactive { + false => Formatter::print(&report, ¶ms), + true => Interactive::init(&report.groups, ¶ms)?, + } Ok(()) } diff --git a/src/pipeline.rs b/src/pipeline.rs new file mode 100644 index 0000000..914243f --- /dev/null +++ b/src/pipeline.rs @@ -0,0 +1,97 @@ +use anyhow::Result; +use indicatif::{ProgressBar, ProgressStyle}; +use rand::Rng; +use std::time::Duration; +use unicode_segmentation::UnicodeSegmentation; + +use crate::fileinfo::FileInfo; +use crate::params::Params; +use crate::processor; +use crate::scanner::Scanner; + +pub struct DuplicateGroup { + pub hash: u128, + pub files: Vec, +} + +pub struct DedupReport { + pub groups: Vec, + pub max_path_len: usize, +} + +pub(crate) fn spinner(enabled: bool, message: &'static str) -> ProgressBar { + let bar = if enabled { + ProgressBar::new_spinner() + } else { + ProgressBar::hidden() + }; + + let style = ProgressStyle::with_template("[{elapsed_precise}] {pos:>7} {msg}") + .expect("valid progress template"); + + bar.set_style(style); + bar.enable_steady_tick(Duration::from_millis(50)); + bar.set_message(message); + bar +} + +pub fn run(params: &Params) -> Result { + let seed: i64 = rand::rng().random(); + + let files = Scanner::new(params)?.scan()?; + let size_groups = processor::group_by_size(files, params.progress); + + let candidates: Vec = size_groups + .into_iter() + .filter(|group| group.len() > 1) + .flatten() + .collect(); + + let max_path_len = candidates + .iter() + .map(|file| file.path.to_string_lossy().graphemes(true).count()) + .max() + .unwrap_or(0); + + let groups = processor::group_by_hash(candidates, params.strict, seed, params.progress); + + Ok(DedupReport { + groups, + max_path_len, + }) +} + +#[cfg(test)] +mod tests { + use super::run; + use crate::params::Params; + use anyhow::Result; + use std::fs::File; + use std::io::Write; + use tempfile::TempDir; + + #[test] + fn run_finds_a_single_group_of_identical_files() -> Result<()> { + let root = TempDir::new()?; + + let duplicate = vec![7u8; 200_000]; + for name in ["a.bin", "b.bin"] { + let mut file = File::create_new(root.path().join(name))?; + file.write_all(&duplicate)?; + } + + let mut unique = File::create_new(root.path().join("c.bin"))?; + unique.write_all(&vec![9u8; 100_000])?; + + let params = Params { + dir: Some(root.path().into()), + ..Default::default() + }; + + let report = run(¶ms)?; + + assert_eq!(report.groups.len(), 1); + assert_eq!(report.groups[0].files.len(), 2); + Ok(()) + } +} diff --git a/src/processor.rs b/src/processor.rs index 1c20a3b..f848ebe 100644 --- a/src/processor.rs +++ b/src/processor.rs @@ -1,382 +1,171 @@ -use anyhow::Result; -use dashmap::DashMap; -use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; -use rayon::iter::IntoParallelRefMutIterator; +use std::collections::HashMap; + use rayon::prelude::{IntoParallelIterator, ParallelIterator}; -use std::sync::atomic::{AtomicBool, AtomicU64, Ordering}; -use std::sync::{Arc, Mutex, TryLockError, TryLockResult}; -use std::time::Duration; -use unicode_segmentation::UnicodeSegmentation; use crate::fileinfo::FileInfo; -use crate::params::Params; +use crate::pipeline::{spinner, DuplicateGroup}; -pub struct Processor {} +pub fn group_by_size(files: Vec, progress: bool) -> Vec> { + let bar = spinner(progress, "files grouped by size"); -impl Processor { - pub fn hashwise( - app_args: Arc, - sw_store: Arc>>, - hw_store: Arc>>, - 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 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."); - - 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(); - - 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(); - - let fhash = match app_args.strict { - true => file.hash(seed).expect("hashing file failed."), - false => file.initpages_hash(seed).expect("hashing file failed."), - }; - - Self::compare_and_update_max_path_len( - max_file_size.clone(), - file.path.to_string_lossy().graphemes(true).count() as u64, - ); - - hw_store - .entry(fhash) - .and_modify(|fileset| fileset.push(file.clone())) - .or_insert_with(|| vec![file.clone()]); - }); - }; - }); - } - } + let mut buckets: HashMap> = HashMap::new(); + for file in files { + bar.inc(1); + buckets.entry(file.size).or_default().push(file); } - pub fn compare_and_update_max_path_len(current: Arc, next: u64) { - if current.load(Ordering::Relaxed) < next { - current.store(next, Ordering::Release); - } - } + bar.finish_with_message("files grouped by size"); + buckets.into_values().collect() +} - pub fn sizewise( - app_args: Arc, - scanner_finished: Arc, - store: Arc>>, - files: Arc>>, - progress_bar_box: Arc, - ) -> Result<()> { - let progress_bar = match app_args.progress { - true => progress_bar_box.add(ProgressBar::new_spinner()), - false => ProgressBar::hidden(), - }; +pub fn group_by_hash( + candidates: Vec, + strict: bool, + seed: i64, + progress: bool, +) -> Vec { + let bar = spinner(progress, "files grouped by hash"); - 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 size"); - - loop { - let fileopt: Option = { - match files.try_lock() { - Ok(mut flist) => flist.pop(), - TryLockResult::Err(TryLockError::WouldBlock) => None, - _ => None, - } + let hashed: Vec<(u128, FileInfo)> = candidates + .into_par_iter() + .map(|file| { + bar.inc(1); + let hash = match strict { + true => file.hash(seed).expect("hashing file failed."), + false => file.initpages_hash(seed).expect("hashing file failed."), }; + (hash, file) + }) + .collect(); - match fileopt { - Some(file) => { - progress_bar.inc(1); - 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 => { - progress_bar.finish_with_message("files grouped by size"); - break Ok(()); - } - false => continue, - }, - } - } + bar.finish_with_message("files grouped by hash."); + + let mut buckets: HashMap> = HashMap::new(); + for (hash, file) in hashed { + buckets.entry(hash).or_default().push(file); } + + buckets + .into_iter() + .filter(|(_, files)| files.len() > 1) + .map(|(hash, files)| DuplicateGroup { hash, files }) + .collect() } #[cfg(test)] -mod tests { +mod staged_tests { use anyhow::Result; - use dashmap::DashMap; - use indicatif::MultiProgress; use rand::Rng; use std::fs::File; use std::io::Write; - use std::sync::atomic::{AtomicBool, AtomicU64}; - use std::sync::{Arc, Mutex}; use tempfile::TempDir; - use crate::{fileinfo::FileInfo, params::Params}; - - use super::Processor; + use crate::fileinfo::FileInfo; fn generate_bytes(size: usize) -> Vec { let mut rng = rand::rng(); (0..size).map(|_| rng.random::()).collect::>() } + fn write_files(root: &TempDir, specs: Vec<(&str, Vec)>) -> Result> { + specs + .into_iter() + .map(|(name, content)| { + let path = root.path().join(name); + let mut file = File::create_new(&path)?; + file.write_all(&content)?; + FileInfo::new(path) + }) + .collect() + } + #[test] - fn hashwise_sorting_two_files_with_identical_init_pages_only_strict_mode() -> Result<()> { + fn group_by_size_separates_files_of_different_sizes() -> Result<()> { let root = TempDir::new()?; - let content = generate_bytes(16384); - - let mut content_x = content.clone(); - let mut content_y = content.clone(); - - content_x.extend(generate_bytes(1720320)); - content_y.extend(generate_bytes(1720320)); - - let files = [ - (root.path().join("fileone.bin"), content_x), - (root.path().join("filetwo.bin"), content_y), - ]; - - for (fpath, content) in files.iter() { - let mut f = File::create_new(fpath)?; - f.write_all(content)?; - } - - let dupstore = Arc::new(DashMap::new()); - let file_queue = Arc::new(Mutex::new( - files - .iter() - .map(|f| FileInfo::new(f.0.clone()).unwrap()) - .collect::>(), - )); - - let hw_dupstore = Arc::new(DashMap::new()); - Processor::sizewise( - Arc::new(Params::default()), - Arc::new(AtomicBool::new(true)), - dupstore.clone(), - file_queue, - Arc::new(MultiProgress::new()), + let files = write_files( + &root, + vec![ + ("fileone.bin", generate_bytes(282624)), + ("filetwo.bin", generate_bytes(1720320)), + ], )?; - let args = Params { - strict: true, - ..Default::default() - }; - - Processor::hashwise( - Arc::new(args), - dupstore.clone(), - hw_dupstore.clone(), - Arc::new(MultiProgress::new()), - Arc::new(AtomicU64::new(32)), - 300, - Arc::new(AtomicBool::new(true)), - )?; - - assert_eq!(hw_dupstore.len(), 2); - + let groups = super::group_by_size(files, false); + assert_eq!(groups.len(), 2); Ok(()) } #[test] - fn hashwise_sorting_two_files_with_identical_init_pages_only_fast_mode() -> Result<()> { + fn group_by_size_buckets_same_size_files_together() -> Result<()> { let root = TempDir::new()?; - let content = generate_bytes(16384); - - let mut content_x = content.clone(); - let mut content_y = content.clone(); - - content_x.extend(generate_bytes(1720320)); - content_y.extend(generate_bytes(1720320)); - - let files = [ - (root.path().join("fileone.bin"), content_x), - (root.path().join("filetwo.bin"), content_y), - ]; - - for (fpath, content) in files.iter() { - let mut f = File::create_new(fpath)?; - f.write_all(content)?; - } - - let dupstore = Arc::new(DashMap::new()); - let file_queue = Arc::new(Mutex::new( - files - .iter() - .map(|f| FileInfo::new(f.0.clone()).unwrap()) - .collect::>(), - )); - - let hw_dupstore = Arc::new(DashMap::new()); - Processor::sizewise( - Arc::new(Params::default()), - Arc::new(AtomicBool::new(true)), - dupstore.clone(), - file_queue, - Arc::new(MultiProgress::new()), + let files = write_files( + &root, + vec![ + ("fileone.bin", generate_bytes(282624)), + ("filetwo.bin", generate_bytes(282624)), + ], )?; - Processor::hashwise( - Arc::new(Params::default()), - dupstore.clone(), - hw_dupstore.clone(), - Arc::new(MultiProgress::new()), - Arc::new(AtomicU64::new(32)), - 300, - Arc::new(AtomicBool::new(true)), - )?; - - assert_eq!(hw_dupstore.len(), 1); - + let groups = super::group_by_size(files, false); + assert_eq!(groups.len(), 1); Ok(()) } #[test] - fn hashwise_sorting_two_files_with_identical_data() -> Result<()> { + fn group_by_hash_fast_mode_matches_identical_init_pages() -> Result<()> { + let root = TempDir::new()?; + let shared = generate_bytes(16384); + + let mut content_x = shared.clone(); + let mut content_y = shared.clone(); + content_x.extend(generate_bytes(1720320)); + content_y.extend(generate_bytes(1720320)); + + let files = write_files( + &root, + vec![("fileone.bin", content_x), ("filetwo.bin", content_y)], + )?; + + let groups = super::group_by_hash(files, false, 300, false); + assert_eq!(groups.len(), 1); + Ok(()) + } + + #[test] + fn group_by_hash_strict_mode_rejects_different_tails() -> Result<()> { + let root = TempDir::new()?; + let shared = generate_bytes(16384); + + let mut content_x = shared.clone(); + let mut content_y = shared.clone(); + content_x.extend(generate_bytes(1720320)); + content_y.extend(generate_bytes(1720320)); + + let files = write_files( + &root, + vec![("fileone.bin", content_x), ("filetwo.bin", content_y)], + )?; + + let groups = super::group_by_hash(files, true, 300, false); + assert_eq!(groups.len(), 0); + Ok(()) + } + + #[test] + fn group_by_hash_matches_identical_files() -> Result<()> { let root = TempDir::new()?; let content = generate_bytes(282624); - let files = [ - (root.path().join("fileone.bin"), content.clone()), - (root.path().join("filetwo.bin"), content.clone()), - ]; - for (fpath, content) in files.iter() { - let mut f = File::create_new(fpath)?; - f.write_all(content)?; - } - - let dupstore = Arc::new(DashMap::new()); - let file_queue = Arc::new(Mutex::new( - files - .iter() - .map(|f| FileInfo::new(f.0.clone()).unwrap()) - .collect::>(), - )); - - let hw_dupstore = Arc::new(DashMap::new()); - Processor::sizewise( - Arc::new(Params::default()), - Arc::new(AtomicBool::new(true)), - dupstore.clone(), - file_queue, - Arc::new(MultiProgress::new()), + let files = write_files( + &root, + vec![ + ("fileone.bin", content.clone()), + ("filetwo.bin", content.clone()), + ], )?; - Processor::hashwise( - Arc::new(Params::default()), - dupstore.clone(), - hw_dupstore.clone(), - Arc::new(MultiProgress::new()), - Arc::new(AtomicU64::new(32)), - 300, - Arc::new(AtomicBool::new(true)), - )?; - - assert_eq!(hw_dupstore.len(), 1); - - Ok(()) - } - - #[test] - fn sizewise_sorting_two_files_of_different_sizes() -> Result<()> { - let root = TempDir::new()?; - let files = [ - (root.path().join("fileone.bin"), generate_bytes(282624)), - (root.path().join("filetwo.bin"), generate_bytes(1720320)), - ]; - - for (fpath, content) in files.iter() { - let mut f = File::create_new(fpath)?; - f.write_all(content)?; - } - - let file_queue = Arc::new(Mutex::new( - files - .iter() - .map(|f| FileInfo::new(f.0.clone()).unwrap()) - .collect::>(), - )); - - let dupstore = Arc::new(DashMap::new()); - - Processor::sizewise( - Arc::new(Params::default()), - Arc::new(AtomicBool::new(true)), - dupstore.clone(), - file_queue, - Arc::new(MultiProgress::new()), - )?; - - assert_eq!(dupstore.len(), 2); - - Ok(()) - } - - #[test] - fn sizewise_sorting_two_files_of_same_size() -> Result<()> { - let root = TempDir::new()?; - let files = [ - (root.path().join("fileone.bin"), generate_bytes(282624)), - (root.path().join("filetwo.bin"), generate_bytes(282624)), - ]; - - for (fpath, content) in files.iter() { - let mut f = File::create_new(fpath)?; - f.write_all(content)?; - } - - let file_queue = Arc::new(Mutex::new( - files - .iter() - .map(|f| FileInfo::new(f.0.clone()).unwrap()) - .collect::>(), - )); - - let dupstore = Arc::new(DashMap::new()); - - Processor::sizewise( - Arc::new(Params::default()), - Arc::new(AtomicBool::new(true)), - dupstore.clone(), - file_queue, - Arc::new(MultiProgress::new()), - )?; - - assert_eq!(dupstore.len(), 1); - + let groups = super::group_by_hash(files, false, 300, false); + assert_eq!(groups.len(), 1); Ok(()) } } diff --git a/src/scanner.rs b/src/scanner.rs index f2e690b..fb467dd 100644 --- a/src/scanner.rs +++ b/src/scanner.rs @@ -1,9 +1,6 @@ -use crate::{fileinfo::FileInfo, params::Params}; +use crate::{fileinfo::FileInfo, params::Params, pipeline::spinner}; use anyhow::Result; -use indicatif::{MultiProgress, ProgressBar, ProgressStyle}; -use std::sync::{Arc, Mutex}; -use std::{path::Path, time::Duration}; - +use std::path::Path; use globwalk::{GlobWalker, GlobWalkerBuilder}; pub struct Scanner { @@ -18,7 +15,7 @@ pub struct Scanner { } impl Scanner { - pub fn new(app_args: Arc) -> Result { + pub fn new(app_args: &Params) -> Result { Ok(Self { directory: app_args.get_directory()?.into_boxed_path(), include_types: app_args.types.clone(), @@ -65,6 +62,7 @@ impl Scanner { None => Ok(walker), } } + fn build_walker(&self) -> Result { let walker = Ok(GlobWalkerBuilder::from_patterns( self.directory.clone(), @@ -77,50 +75,32 @@ impl Scanner { Ok(walker.build()?) } - pub fn scan( - &self, - files: Arc>>, - progress_bar_box: Arc, - ) -> Result<()> { - let progress_bar = match self.progress { - true => progress_bar_box.add(ProgressBar::new_spinner()), - false => ProgressBar::hidden(), - }; - - 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("paths mapped"); + pub fn scan(&self) -> Result> { + let bar = spinner(self.progress, "paths mapped"); let min_size = self.min_size.unwrap_or(0); - self.build_walker()? + let files: Vec = self + .build_walker()? .filter_map(Result::ok) .map(|entity| entity.into_path()) - .inspect(|_path| progress_bar.inc(1)) + .inspect(|_path| bar.inc(1)) .filter(|path| path.is_file()) .map(FileInfo::new) .filter_map(Result::ok) .filter(|file| file.size >= min_size) - .for_each(|file| { - let mut flock = files.lock().unwrap(); - flock.push(file); - }); + .collect(); - progress_bar.finish_with_message("paths mapped"); - Ok(()) + bar.finish_with_message("paths mapped"); + Ok(files) } } #[cfg(test)] mod tests { - use crate::fileinfo::FileInfo; use crate::params::Params; use std::fs::File; - use std::sync::{Arc, Mutex}; - - use super::Scanner; - use indicatif::MultiProgress; use tempfile::TempDir; + use super::Scanner; #[test] fn ensure_file_include_type_filter_includes_expected_file_types() { @@ -145,26 +125,16 @@ mod tests { ..Default::default() }; - let progress = Arc::new(MultiProgress::new()); - let scanlist = Arc::new(Mutex::>::new(vec![])); - let scanner = Scanner::new(Arc::new(params)).expect("scanner initialization failed"); + let scanner = Scanner::new(¶ms).expect("scanner initialization failed"); + let files = scanner.scan().expect("scanning failed."); - scanner - .scan(scanlist.clone(), progress) - .expect("scanning failed."); - - let scan_list_mg = scanlist.lock().unwrap(); - - assert!(scan_list_mg.iter().any(|f| f.path.to_str().unwrap() + assert!(files.iter().any(|f| f.path.to_str().unwrap() == root.path().join("this-is-a-js-file.js").to_str().unwrap())); - - assert!(scan_list_mg.iter().any(|f| f.path.to_str().unwrap() + assert!(files.iter().any(|f| f.path.to_str().unwrap() == root.path().join("this-is-a-csv-file.csv").to_str().unwrap())); - - assert!(scan_list_mg.iter().all(|f| f.path.to_str().unwrap() + assert!(files.iter().all(|f| f.path.to_str().unwrap() != root.path().join("this-is-a-css-file.css").to_str().unwrap())); - - assert!(scan_list_mg.iter().all(|f| f.path.to_str().unwrap() + assert!(files.iter().all(|f| f.path.to_str().unwrap() != root.path().join("this-is-a-rust-file.rs").to_str().unwrap())); } @@ -191,26 +161,19 @@ mod tests { ..Default::default() }; - let progress = Arc::new(MultiProgress::new()); - let scanlist = Arc::new(Mutex::>::new(vec![])); - let scanner = Scanner::new(Arc::new(params)).expect("scanner initialization failed"); + let scanner = Scanner::new(¶ms).expect("scanner initialization failed"); + let files = scanner.scan().expect("scanning failed."); - scanner - .scan(scanlist.clone(), progress) - .expect("scanning failed."); - - let scan_list_mg = scanlist.lock().unwrap(); - - assert!(scan_list_mg.iter().all(|f| f.path.to_str().unwrap() + assert!(files.iter().all(|f| f.path.to_str().unwrap() != root.path().join("this-is-a-js-file.js").to_str().unwrap())); - assert!(scan_list_mg.iter().all(|f| f.path.to_str().unwrap() + assert!(files.iter().all(|f| f.path.to_str().unwrap() != root.path().join("this-is-a-csv-file.csv").to_str().unwrap())); - assert!(scan_list_mg.iter().any(|f| f.path.to_str().unwrap() + assert!(files.iter().any(|f| f.path.to_str().unwrap() == root.path().join("this-is-a-css-file.css").to_str().unwrap())); - assert!(scan_list_mg.iter().any(|f| f.path.to_str().unwrap() + assert!(files.iter().any(|f| f.path.to_str().unwrap() == root.path().join("this-is-a-rust-file.rs").to_str().unwrap())); } @@ -238,23 +201,16 @@ mod tests { ..Default::default() }; - let progress = Arc::new(MultiProgress::new()); - let scanlist = Arc::new(Mutex::>::new(vec![])); - let scanner = Scanner::new(Arc::new(params)).expect("scanner initialization failed"); + let scanner = Scanner::new(¶ms).expect("scanner initialization failed"); + let files = scanner.scan().expect("scanning failed."); - scanner - .scan(scanlist.clone(), progress) - .expect("scanning failed."); - - let scan_list_mg = scanlist.lock().unwrap(); - - assert!(scan_list_mg.iter().any(|f| f.path.to_str().unwrap() + assert!(files.iter().any(|f| f.path.to_str().unwrap() == root.path().join("this-is-a-js-file.js").to_str().unwrap())); - assert!(scan_list_mg.iter().all(|f| f.path.to_str().unwrap() + assert!(files.iter().all(|f| f.path.to_str().unwrap() != root.path().join("this-is-a-csv-file.csv").to_str().unwrap())); - assert!(scan_list_mg.iter().any(|f| f.path.to_str().unwrap() + assert!(files.iter().any(|f| f.path.to_str().unwrap() == root.path().join("this-is-a-rust-file.rs").to_str().unwrap())); } } diff --git a/src/server.rs b/src/server.rs deleted file mode 100644 index 539a522..0000000 --- a/src/server.rs +++ /dev/null @@ -1,117 +0,0 @@ -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 indicatif::{MultiProgress, ProgressDrawTarget}; -use rand::Rng; -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(4), - app_args: Arc::new(opts), - max_file_path_len: Arc::new(AtomicU64::new(0)), - } - } - - pub fn start(&self) -> Result<()> { - let progbarbox = Arc::new(MultiProgress::new()); - let mut rng = rand::rng(); - let seed: i64 = rng.random(); - - if !self.app_args.progress { - progbarbox.set_draw_target(ProgressDrawTarget::hidden()); - } - - let (app_args_sc, app_args_sw, app_args_hw) = ( - Arc::clone(&self.app_args), - Arc::clone(&self.app_args), - Arc::clone(&self.app_args), - ); - let (file_queue_sc, file_queue_pr) = ( - Arc::clone(&self.filequeue), - Arc::clone(&self.filequeue), - ); - let scanner_finished = Arc::new(AtomicBool::new(false)); - let sw_sort_finished = Arc::new(AtomicBool::new(false)); - let (sfin_sc, sfin_pr) = ( - Arc::clone(&scanner_finished), - Arc::clone(&scanner_finished), - ); - let (swfin_pr_sw, swfin_pr_hw) = ( - Arc::clone(&sw_sort_finished), - Arc::clone(&sw_sort_finished), - ); - let (store_sw, store_sw2, store_hw) = ( - Arc::clone(&self.sw_duplicate_set), - Arc::clone(&self.sw_duplicate_set), - Arc::clone(&self.hw_duplicate_set), - ); - let max_file_path_len = Arc::clone(&self.max_file_path_len); - let (prog_sc, prog_sw, prog_hw) = ( - Arc::clone(&progbarbox), - Arc::clone(&progbarbox), - Arc::clone(&progbarbox), - ); - - self.threadpool.execute(move || { - Scanner::new(app_args_sc) - .expect("unable to initialize scanner.") - .scan(file_queue_sc, prog_sc) - .expect("scanner failed."); - - sfin_sc.store(true, std::sync::atomic::Ordering::Relaxed); - }); - - self.threadpool.execute(move || { - Processor::sizewise( - app_args_sw, - sfin_pr, - store_sw, - file_queue_pr, - prog_sw, - ) - .expect("sizewise scanner failed."); - - swfin_pr_sw.store(true, std::sync::atomic::Ordering::Relaxed); - }); - - self.threadpool.execute(move || { - Processor::hashwise( - app_args_hw, - store_sw2, - store_hw, - prog_hw, - max_file_path_len, - seed, - swfin_pr_hw, - ) - .expect("sizewise scanner failed."); - }); - - progbarbox.clear()?; - - self.threadpool.join(); - - Ok(()) - } -}