Parallel working
This commit is contained in:
parent
464fc6e15a
commit
8ee4a67bbe
3 changed files with 124 additions and 68 deletions
90
Cargo.lock
generated
90
Cargo.lock
generated
|
|
@ -68,6 +68,28 @@ dependencies = [
|
|||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam"
|
||||
version = "0.8.4"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1137cd7e7fc0fb5d3c5a8678be38ec56e819125d8d7907411fe24ccb943faca8"
|
||||
dependencies = [
|
||||
"crossbeam-channel",
|
||||
"crossbeam-deque",
|
||||
"crossbeam-epoch",
|
||||
"crossbeam-queue",
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-channel"
|
||||
version = "0.5.13"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "33480d6946193aa8033910124896ca395333cae7e2d1113d1fef6c3272217df2"
|
||||
dependencies = [
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-deque"
|
||||
version = "0.8.5"
|
||||
|
|
@ -87,6 +109,15 @@ dependencies = [
|
|||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-queue"
|
||||
version = "0.3.11"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "df0346b5d5e76ac2fe4e327c5fd1118d6be7c51dfb18f9b7922923f287471e35"
|
||||
dependencies = [
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "crossbeam-utils"
|
||||
version = "0.8.20"
|
||||
|
|
@ -139,10 +170,13 @@ dependencies = [
|
|||
]
|
||||
|
||||
[[package]]
|
||||
name = "either"
|
||||
version = "1.12.0"
|
||||
name = "dpc-pariter"
|
||||
version = "0.5.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3dca9240753cf90908d7e4aac30f630662b02aebaa1b58a3cadabdb23385b58b"
|
||||
checksum = "1a6f60e0061d01135c7e9b77486a7f5a8e63ba14acbdbca73e42007cfd8a1c91"
|
||||
dependencies = [
|
||||
"pariter",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "fallible-iterator"
|
||||
|
|
@ -185,10 +219,10 @@ dependencies = [
|
|||
]
|
||||
|
||||
[[package]]
|
||||
name = "hex-literal"
|
||||
version = "0.4.1"
|
||||
name = "hermit-abi"
|
||||
version = "0.3.9"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "6fe2267d4ed49bc07b63801559be28c718ea06c4738b7a03c94df7386d2cde46"
|
||||
checksum = "d231dfb89cfffdbc30e7fc41579ed6066ad03abda9e567ccafae602b97ec5024"
|
||||
|
||||
[[package]]
|
||||
name = "libc"
|
||||
|
|
@ -235,12 +269,33 @@ dependencies = [
|
|||
"windows-sys 0.48.0",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "num_cpus"
|
||||
version = "1.16.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "4161fcb6d602d4d2081af7c3a45852d875a03dd337a6bfdd6e06407b61342a43"
|
||||
dependencies = [
|
||||
"hermit-abi",
|
||||
"libc",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "once_cell"
|
||||
version = "1.19.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "3fdb12b2476b595f9358c5161aa467c2438859caa136dec86c26fdd2efe17b92"
|
||||
|
||||
[[package]]
|
||||
name = "pariter"
|
||||
version = "0.5.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "324a62b9e7b5f270c0acc92a2040f8028bb643f959f9c068f11a7864f327e3d9"
|
||||
dependencies = [
|
||||
"crossbeam",
|
||||
"crossbeam-channel",
|
||||
"num_cpus",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "parking_lot"
|
||||
version = "0.12.3"
|
||||
|
|
@ -283,8 +338,7 @@ dependencies = [
|
|||
name = "quiksearch"
|
||||
version = "0.1.0"
|
||||
dependencies = [
|
||||
"hex-literal",
|
||||
"rayon",
|
||||
"dpc-pariter",
|
||||
"rusqlite",
|
||||
"sha2",
|
||||
"tqdm",
|
||||
|
|
@ -300,26 +354,6 @@ dependencies = [
|
|||
"proc-macro2",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rayon"
|
||||
version = "1.10.0"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "b418a60154510ca1a002a752ca9714984e21e4241e804d32555251faf8b78ffa"
|
||||
dependencies = [
|
||||
"either",
|
||||
"rayon-core",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "rayon-core"
|
||||
version = "1.12.1"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "1465873a3dfdaa8ae7cb14b4383657caab0b3e8a0aa9ae8e04b044854c8dfce2"
|
||||
dependencies = [
|
||||
"crossbeam-deque",
|
||||
"crossbeam-utils",
|
||||
]
|
||||
|
||||
[[package]]
|
||||
name = "redox_syscall"
|
||||
version = "0.5.1"
|
||||
|
|
|
|||
|
|
@ -4,8 +4,7 @@ version = "0.1.0"
|
|||
edition = "2021"
|
||||
|
||||
[dependencies]
|
||||
hex-literal = "0.4.1"
|
||||
rayon = "1.10.0"
|
||||
dpc-pariter = "0.5.1"
|
||||
rusqlite = { version = "0.31.0", features = ["bundled"] }
|
||||
sha2 = "0.10.8"
|
||||
tqdm = "0.7.0"
|
||||
|
|
|
|||
99
src/main.rs
99
src/main.rs
|
|
@ -1,30 +1,21 @@
|
|||
// use std::borrow::Borrow;
|
||||
use std::ffi::OsString;
|
||||
use std::time::Instant;
|
||||
// use std::path::Path;
|
||||
use std::fs::File;
|
||||
use std::io::{Read, Seek, SeekFrom};
|
||||
use std::time::UNIX_EPOCH;
|
||||
use std::sync::{Mutex, Arc};
|
||||
|
||||
use sha2::{Sha256, Digest};
|
||||
use walkdir::WalkDir;
|
||||
use walkdir::{WalkDir, DirEntry};
|
||||
use tqdm;
|
||||
use rusqlite::{params, Connection};
|
||||
|
||||
use dpc_pariter::IteratorExt as _;
|
||||
|
||||
const HASHLEN:usize = 1024*1;
|
||||
|
||||
// struct Finfo {
|
||||
// name: String,
|
||||
// path: String,
|
||||
// size: u64,
|
||||
// modified: u64,
|
||||
// hash: [u8; 64]
|
||||
// }
|
||||
|
||||
fn get_file_hash(size: u64, path: OsString) -> Result<Vec<u8>, std::io::Error> {
|
||||
let mut hasher = Sha256::new();
|
||||
let mut f = File::open(path)?;
|
||||
let mut f: File = File::open(path)?;
|
||||
let mut data = [0u8; 8+2*HASHLEN];
|
||||
data[..8].copy_from_slice(&size.to_le_bytes());
|
||||
if size > HASHLEN as u64 {
|
||||
|
|
@ -48,22 +39,67 @@ fn get_file_hash(size: u64, path: OsString) -> Result<Vec<u8>, std::io::Error> {
|
|||
data[8+i] = file_block[i];
|
||||
}
|
||||
}
|
||||
drop(f);
|
||||
hasher.update(data);
|
||||
Ok(hasher.finalize().to_vec())
|
||||
|
||||
}
|
||||
|
||||
fn process_entry(conn_mutex: &Arc<Mutex<Connection>>, entry: DirEntry) {
|
||||
let meta = entry.metadata().unwrap();
|
||||
if !meta.is_dir() {
|
||||
// let fpath = entry.path().canonicalize()?.into_os_string();
|
||||
let fpath_result = entry.path().canonicalize();
|
||||
let fpath = match fpath_result {
|
||||
Ok(fp) => fp.into_os_string(),
|
||||
Err(error) => {
|
||||
println!("Error converting fpath: {:?}", error);
|
||||
return;
|
||||
},
|
||||
};
|
||||
|
||||
let fsize = meta.len();
|
||||
let fmodified = meta.modified().unwrap().duration_since(UNIX_EPOCH).unwrap().as_secs();
|
||||
|
||||
// let fhash = get_file_hash(fsize, fpath.clone())?;
|
||||
let fhash_result = get_file_hash(fsize, fpath.clone());
|
||||
let fhash = match fhash_result {
|
||||
Ok(fh) => fh,
|
||||
Err(error) => {
|
||||
println!("Error digesting hash: {:?}", error);
|
||||
return;
|
||||
},
|
||||
};
|
||||
// let fhash = b"";
|
||||
|
||||
let query = "INSERT INTO files VALUES (?1,?2,?3,?4,?5)";
|
||||
let conn = conn_mutex.lock().unwrap();
|
||||
let mut stmt = conn.prepare_cached(query).unwrap();
|
||||
let fname = entry.path().file_name().unwrap().to_os_string();
|
||||
|
||||
// stmt.execute(params![fname.to_str(), fpath.to_str(), fsize, fmodified, fhash])?;
|
||||
let stmt_result = stmt.execute(params![fname.to_str(), fpath.to_str(), fsize, fmodified, fhash]);
|
||||
match stmt_result {
|
||||
Ok(us) => us,
|
||||
Err(error) => {
|
||||
println!("Error with sqlite transaction: {:?}", error);
|
||||
return;
|
||||
},
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
fn main() {
|
||||
let path: &str = "Y:\\";
|
||||
let path: &str = "G:\\";
|
||||
let db_path: &str = "YDrive.db";
|
||||
|
||||
let conn = Connection::open(db_path).unwrap();
|
||||
// let conn = Connection::open_in_memory().unwrap();
|
||||
// PRAGMA cache_size is in number of pages with 1024 byte page size by default
|
||||
conn.execute_batch(
|
||||
"PRAGMA journal_mode = OFF;
|
||||
PRAGMA synchronous = 0;
|
||||
PRAGMA cache_size = 1000000;
|
||||
PRAGMA cache_size = 10000;
|
||||
PRAGMA locking_mode = EXCLUSIVE;
|
||||
PRAGMA temp_store = MEMORY;",
|
||||
)
|
||||
|
|
@ -75,29 +111,16 @@ fn main() {
|
|||
moddate INTEGER,
|
||||
hash BLOB)", ()).unwrap();
|
||||
|
||||
let mut filecount: i64 = 0;
|
||||
let start_time = Instant::now();
|
||||
let conn_mutex = Arc::new(Mutex::new(conn));
|
||||
|
||||
for entry in tqdm::tqdm(WalkDir::new(path).into_iter().filter_map(|e| e.ok())) {
|
||||
let meta = entry.metadata().unwrap();
|
||||
if !meta.is_dir() {
|
||||
let query = "INSERT INTO files VALUES (?1,?2,?3,?4,?5)";
|
||||
let mut stmt = conn.prepare_cached(query).unwrap();
|
||||
let fname = entry.path().file_name().unwrap().to_os_string();
|
||||
let fpath = entry.path().canonicalize().unwrap().into_os_string();
|
||||
let fsize = meta.len();
|
||||
let fmodified = meta.modified().unwrap().duration_since(UNIX_EPOCH).unwrap().as_secs();
|
||||
// let fhash = get_file_hash(fsize, fpath.clone()).unwrap();
|
||||
let fhash = b"";
|
||||
filecount += 1;
|
||||
stmt.execute(params![fname.to_str(), fpath.to_str(), fsize, fmodified, fhash]).unwrap();
|
||||
}
|
||||
}
|
||||
let elapsed_time = start_time.elapsed();
|
||||
println!("{} files enumerated in {} seconds", filecount, elapsed_time.as_secs());
|
||||
println!("{} files per second", filecount as f32 / elapsed_time.as_millis() as f32 * 1000.0);
|
||||
|
||||
// select name, hash, count(hash) as cnt from files group by hash
|
||||
// ORDER BY cnt DESC;
|
||||
tqdm::tqdm(WalkDir::new(path).into_iter()).parallel_map(move |entry| {
|
||||
match entry {
|
||||
Ok(e) => process_entry(&conn_mutex, e),
|
||||
Err(error) => {
|
||||
println!("Error with directory walk: {:?}", error);
|
||||
return;
|
||||
},
|
||||
};
|
||||
}).for_each(drop);
|
||||
|
||||
}
|
||||
Loading…
Add table
Reference in a new issue