mirror of
https://github.com/sreedevk/deduplicator.git
synced 2026-08-26 10:05:32 +00:00
[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<Mutex<Vec>> hand-off queue, DashMap stores, and a per-file Arc<Mutex<FileState>>. 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.
This commit is contained in:
Sreedev Kodichath
3
.gitignore
vendored
3
.gitignore
vendored
@@ -2,3 +2,6 @@
|
||||
/Cargo.lock
|
||||
/result-bin
|
||||
/.bacon-locations
|
||||
/docs
|
||||
/.claude
|
||||
/.superpowers
|
||||
|
||||
86
Cargo.lock
generated
86
Cargo.lock
generated
@@ -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"
|
||||
|
||||
@@ -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]
|
||||
|
||||
@@ -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<Path>,
|
||||
pub size: u64,
|
||||
pub modified: SystemTime,
|
||||
pub state: Arc<Mutex<FileState>>,
|
||||
}
|
||||
|
||||
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)]
|
||||
|
||||
@@ -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<DashMap<u128, Vec<FileInfo>>>, 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::<String>();
|
||||
|
||||
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::<String>();
|
||||
|
||||
ostring.push_str(&subfields);
|
||||
println!("{ostring}");
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
@@ -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<DashMap<u128, Vec<FileInfo>>>, 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(())
|
||||
}
|
||||
|
||||
27
src/main.rs
27
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(())
|
||||
}
|
||||
|
||||
97
src/pipeline.rs
Normal file
97
src/pipeline.rs
Normal file
@@ -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<FileInfo>,
|
||||
}
|
||||
|
||||
pub struct DedupReport {
|
||||
pub groups: Vec<DuplicateGroup>,
|
||||
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<DedupReport> {
|
||||
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<FileInfo> = 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(())
|
||||
}
|
||||
}
|
||||
453
src/processor.rs
453
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<FileInfo>, progress: bool) -> Vec<Vec<FileInfo>> {
|
||||
let bar = spinner(progress, "files grouped by size");
|
||||
|
||||
impl Processor {
|
||||
pub fn hashwise(
|
||||
app_args: Arc<Params>,
|
||||
sw_store: Arc<DashMap<u64, Vec<FileInfo>>>,
|
||||
hw_store: Arc<DashMap<u128, Vec<FileInfo>>>,
|
||||
progress_bar_box: Arc<MultiProgress>,
|
||||
max_file_size: Arc<AtomicU64>,
|
||||
seed: i64,
|
||||
sw_sorting_finished: Arc<AtomicBool>,
|
||||
) -> 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<u64> = 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<FileInfo> = 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<u64, Vec<FileInfo>> = 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<AtomicU64>, 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<Params>,
|
||||
scanner_finished: Arc<AtomicBool>,
|
||||
store: Arc<DashMap<u64, Vec<FileInfo>>>,
|
||||
files: Arc<Mutex<Vec<FileInfo>>>,
|
||||
progress_bar_box: Arc<MultiProgress>,
|
||||
) -> 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<FileInfo>,
|
||||
strict: bool,
|
||||
seed: i64,
|
||||
progress: bool,
|
||||
) -> Vec<DuplicateGroup> {
|
||||
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<FileInfo> = {
|
||||
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<u128, Vec<FileInfo>> = 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<u8> {
|
||||
let mut rng = rand::rng();
|
||||
(0..size).map(|_| rng.random::<u8>()).collect::<Vec<u8>>()
|
||||
}
|
||||
|
||||
fn write_files(root: &TempDir, specs: Vec<(&str, Vec<u8>)>) -> Result<Vec<FileInfo>> {
|
||||
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::<Vec<FileInfo>>(),
|
||||
));
|
||||
|
||||
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::<Vec<FileInfo>>(),
|
||||
));
|
||||
|
||||
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::<Vec<FileInfo>>(),
|
||||
));
|
||||
|
||||
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::<Vec<FileInfo>>(),
|
||||
));
|
||||
|
||||
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::<Vec<FileInfo>>(),
|
||||
));
|
||||
|
||||
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(())
|
||||
}
|
||||
}
|
||||
|
||||
104
src/scanner.rs
104
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<Params>) -> Result<Self> {
|
||||
pub fn new(app_args: &Params) -> Result<Self> {
|
||||
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<GlobWalker> {
|
||||
let walker = Ok(GlobWalkerBuilder::from_patterns(
|
||||
self.directory.clone(),
|
||||
@@ -77,50 +75,32 @@ impl Scanner {
|
||||
Ok(walker.build()?)
|
||||
}
|
||||
|
||||
pub fn scan(
|
||||
&self,
|
||||
files: Arc<Mutex<Vec<FileInfo>>>,
|
||||
progress_bar_box: Arc<MultiProgress>,
|
||||
) -> 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<Vec<FileInfo>> {
|
||||
let bar = spinner(self.progress, "paths mapped");
|
||||
let min_size = self.min_size.unwrap_or(0);
|
||||
|
||||
self.build_walker()?
|
||||
let files: Vec<FileInfo> = 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::<Vec<FileInfo>>::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::<Vec<FileInfo>>::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::<Vec<FileInfo>>::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()));
|
||||
}
|
||||
}
|
||||
|
||||
117
src/server.rs
117
src/server.rs
@@ -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<Mutex<Vec<FileInfo>>>,
|
||||
sw_duplicate_set: Arc<DashMap<u64, Vec<FileInfo>>>,
|
||||
pub hw_duplicate_set: Arc<DashMap<u128, 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(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(())
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user