From 53282c50df370614c13bc9b061e216518541d7f4 Mon Sep 17 00:00:00 2001 From: sreedevk Date: Sun, 6 Jul 2025 12:45:06 +0000 Subject: [PATCH] implemented new multi threaded scanner --- Cargo.lock | 115 +++++++++++++++++++++++++++++++++------- Cargo.toml | 2 + src/main.rs | 1 + src/pipeline/file.rs | 42 +++++++++++++++ src/pipeline/mod.rs | 41 ++++++++++++++ src/pipeline/scanner.rs | 85 +++++++++++++++++++++++++++++ 6 files changed, 268 insertions(+), 18 deletions(-) create mode 100644 src/pipeline/file.rs create mode 100644 src/pipeline/mod.rs create mode 100644 src/pipeline/scanner.rs diff --git a/Cargo.lock b/Cargo.lock index ef8ad92..5aa2658 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1,6 +1,6 @@ # This file is automatically @generated by Cargo. # It is not intended for manual editing. -version = 3 +version = 4 [[package]] name = "aho-corasick" @@ -302,7 +302,9 @@ dependencies = [ "rayon", "serde", "serde_json", + "threadpool", "unicode-segmentation", + "uuid", ] [[package]] @@ -352,7 +354,19 @@ checksum = "c4567c8db10ae91089c99af84c68c38da3ec2f087c3f82960bcdbf3656b6f4d7" dependencies = [ "cfg-if", "libc", - "wasi", + "wasi 0.11.0+wasi-snapshot-preview1", +] + +[[package]] +name = "getrandom" +version = "0.3.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "26145e563e54f2cadc477553f1ec5ee650b00862f0a58bcd12cbdc5f0ea2d2f4" +dependencies = [ + "cfg-if", + "libc", + "r-efi", + "wasi 0.14.2+wasi-0.2.4", ] [[package]] @@ -406,6 +420,12 @@ version = "0.3.9" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "d231dfb89cfffdbc30e7fc41579ed6066ad03abda9e567ccafae602b97ec5024" +[[package]] +name = "hermit-abi" +version = "0.5.2" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "fc0fef456e4baa96da950455cd02c081ca953b141298e41db3fc7e36b1da849c" + [[package]] name = "iana-time-zone" version = "0.1.60" @@ -474,7 +494,7 @@ version = "0.4.12" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "f23ff5ef2b80d608d61efee834934d862cd92461afc0560dedf493e4c033738b" dependencies = [ - "hermit-abi", + "hermit-abi 0.3.9", "libc", "windows-sys 0.52.0", ] @@ -502,10 +522,11 @@ checksum = "49f1f14873335454500d59611f1cf4a4b0f786f9ac11f4312a78e4cf2566695b" [[package]] name = "js-sys" -version = "0.3.69" +version = "0.3.77" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "29c15563dc2726973df627357ce0c9ddddbea194836909d655df6a75d2cf296d" +checksum = "1cfaf33c695fc6e08064efbc1f72ec937429614f25eef83af942d0e227c3a28f" dependencies = [ + "once_cell", "wasm-bindgen", ] @@ -571,6 +592,16 @@ dependencies = [ "autocfg", ] +[[package]] +name = "num_cpus" +version = "1.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "91df4bbde75afed763b708b7eee1e8e7651e02d97f6d5dd763e89367e957b23b" +dependencies = [ + "hermit-abi 0.5.2", + "libc", +] + [[package]] name = "number_prefix" version = "0.4.0" @@ -640,6 +671,12 @@ dependencies = [ "proc-macro2", ] +[[package]] +name = "r-efi" +version = "5.3.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "69cdb34c158ceb288df11e18b4bd39de994f6657d83847bdffdbd7f346754b0f" + [[package]] name = "rayon" version = "1.10.0" @@ -675,7 +712,7 @@ version = "0.4.5" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "bd283d9651eeda4b2a83a43c1c91b266c40fd76ecd39a50a8c630ae69dc72891" dependencies = [ - "getrandom", + "getrandom 0.2.15", "libredox", "thiserror", ] @@ -809,6 +846,15 @@ dependencies = [ "syn", ] +[[package]] +name = "threadpool" +version = "1.8.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "d050e60b33d41c19108b32cea32164033a9013fe3b46cbd4457559bfbf77afaa" +dependencies = [ + "num_cpus", +] + [[package]] name = "unicode-ident" version = "1.0.12" @@ -833,6 +879,17 @@ version = "0.2.2" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "06abde3611657adf66d383f00b093d7faecc7fa57071cce2578660c9f1010821" +[[package]] +name = "uuid" +version = "1.17.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "3cf4199d1e5d15ddd86a694e4d0dffa9c323ce759fea589f00fef9d81cc1931d" +dependencies = [ + "getrandom 0.3.3", + "js-sys", + "wasm-bindgen", +] + [[package]] name = "walkdir" version = "2.5.0" @@ -850,24 +907,34 @@ source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "9c8d87e72b64a3b4db28d11ce29237c246188f4f51057d65a7eab63b7987e423" [[package]] -name = "wasm-bindgen" -version = "0.2.92" +name = "wasi" +version = "0.14.2+wasi-0.2.4" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "4be2531df63900aeb2bca0daaaddec08491ee64ceecbee5076636a3b026795a8" +checksum = "9683f9a5a998d873c0d21fcbe3c083009670149a8fab228644b8bd36b2c48cb3" +dependencies = [ + "wit-bindgen-rt", +] + +[[package]] +name = "wasm-bindgen" +version = "0.2.100" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "1edc8929d7499fc4e8f0be2262a241556cfc54a0bea223790e71446f2aab1ef5" dependencies = [ "cfg-if", + "once_cell", + "rustversion", "wasm-bindgen-macro", ] [[package]] name = "wasm-bindgen-backend" -version = "0.2.92" +version = "0.2.100" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "614d787b966d3989fa7bb98a654e369c762374fd3213d212cfc0251257e747da" +checksum = "2f0a0651a5c2bc21487bde11ee802ccaf4c51935d0d3d42a6101f98161700bc6" dependencies = [ "bumpalo", "log", - "once_cell", "proc-macro2", "quote", "syn", @@ -876,9 +943,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro" -version = "0.2.92" +version = "0.2.100" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "a1f8823de937b71b9460c0c34e25f3da88250760bec0ebac694b49997550d726" +checksum = "7fe63fc6d09ed3792bd0897b314f53de8e16568c2b3f7982f468c0bf9bd0b407" dependencies = [ "quote", "wasm-bindgen-macro-support", @@ -886,9 +953,9 @@ dependencies = [ [[package]] name = "wasm-bindgen-macro-support" -version = "0.2.92" +version = "0.2.100" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "e94f17b526d0a461a191c78ea52bbce64071ed5c04c9ffe424dcb38f74171bb7" +checksum = "8ae87ea40c9f689fc23f209965b6fb8a99ad69aeeb0231408be24920604395de" dependencies = [ "proc-macro2", "quote", @@ -899,9 +966,12 @@ dependencies = [ [[package]] name = "wasm-bindgen-shared" -version = "0.2.92" +version = "0.2.100" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "af190c94f2773fdb3729c55b007a722abb5384da03bc0986df4c289bf5567e96" +checksum = "1a05d73b933a847d6cccdda8f838a22ff101ad9bf93e33684f39c1f5f0eece3d" +dependencies = [ + "unicode-ident", +] [[package]] name = "winapi" @@ -1081,3 +1151,12 @@ name = "windows_x86_64_msvc" version = "0.52.6" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "589f6da84c646204747d1270a2a5661ea66ed1cced2631d546fdfb155959f9ec" + +[[package]] +name = "wit-bindgen-rt" +version = "0.39.0" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "6f42320e61fe2cfd34354ecb597f86f413484a798ba44a8ca1165c58d42da6c1" +dependencies = [ + "bitflags 2.6.0", +] diff --git a/Cargo.toml b/Cargo.toml index 1aadc65..5236068 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -30,7 +30,9 @@ prettytable-rs = "0.10.0" rayon = "1.6.1" serde = { version = "1.0.192", features = ["derive"] } serde_json = "1.0.108" +threadpool = "1.8.1" unicode-segmentation = "1.10.0" +uuid = { version = "1.17.0", features = ["v4"] } [profile.release] strip = true diff --git a/src/main.rs b/src/main.rs index e14bf75..62dc181 100644 --- a/src/main.rs +++ b/src/main.rs @@ -4,6 +4,7 @@ mod interactive; mod params; mod processor; mod scanner; +mod pipeline; use anyhow::Result; use clap::Parser; diff --git a/src/pipeline/file.rs b/src/pipeline/file.rs new file mode 100644 index 0000000..a70caf1 --- /dev/null +++ b/src/pipeline/file.rs @@ -0,0 +1,42 @@ +use anyhow::Result; +use std::fs::File; +use std::io::Read; +use std::os::unix::fs::MetadataExt; +use std::sync::Arc; +use uuid::Uuid; + +const PARTIAL_SIZE: u64 = 4096; + +pub struct FileMeta { + id: Uuid, + path: Box, + size: u64, + modtime: i64, + partial: Arc<[u8]>, + full_hash: Arc<[u8]>, +} + +impl FileMeta { + fn create_partial(path: &str) -> Result> { + let mut partial_take = File::open(path)?.take(PARTIAL_SIZE); + let mut partial_buffer = Vec::with_capacity(PARTIAL_SIZE as usize); + partial_take.read_to_end(&mut partial_buffer)?; + + Ok(Arc::from(partial_buffer)) + } + + pub fn new(path: Box) -> Result { + let pstr = path.to_string(); + let filemeta = std::fs::metadata(&pstr)?; + let partial = Self::create_partial(&pstr)?; + + Ok(Self { + id: Uuid::new_v4(), + size: filemeta.size(), + modtime: filemeta.mtime(), + full_hash: Arc::new([]), + path, + partial, + }) + } +} diff --git a/src/pipeline/mod.rs b/src/pipeline/mod.rs new file mode 100644 index 0000000..e86bdd0 --- /dev/null +++ b/src/pipeline/mod.rs @@ -0,0 +1,41 @@ +mod file; +mod scanner; +use std::sync::Arc; +use std::sync::Mutex; + +use self::file::FileMeta; + +const CONCURRENCY: usize = 4; + +pub struct Server { } + +impl Server { + // pub fn new() -> Result { + // Ok(Self { + // index_queue: Arc::new(Mutex::new(vec![])), + // index_tpool: ThreadPool::new(CONCURRENCY), + // process_queue: Arc::new(Mutex::new(vec![])), + // }) + // } + + // fn index(&self) -> Result<()> { + // let mut iq = self.index_queue.lock().unwrap(); + // let mut pq = self.process_queue.lock().unwrap(); + // + // match iq.pop() { + // None => Ok(()), + // Some(QueueElem::File(path)) => { + // pq.push(FileMeta::new(path)?); + // + // Ok(()) + // } + // Some(QueueElem::Directory(path)) => { + // self.index_tpool.execute(move || { + // Scanner::new(&path); + // }); + // + // Ok(()) + // } + // } + // } +} diff --git a/src/pipeline/scanner.rs b/src/pipeline/scanner.rs new file mode 100644 index 0000000..8471f30 --- /dev/null +++ b/src/pipeline/scanner.rs @@ -0,0 +1,85 @@ +use anyhow::{anyhow, Result}; +use std::fs; +use std::sync::atomic::{AtomicU32, Ordering::Relaxed}; +use std::sync::Arc; +use std::sync::Mutex; +use threadpool::ThreadPool; + +pub struct Scanner { + files: Arc>>>, + threadpool: Arc, + proc_count: Arc, + proc_queue: Arc>>>, +} + +impl Scanner { + pub fn new(path: Box) -> Result { + Ok(Self { + files: Arc::new(Mutex::new(vec![])), + threadpool: Arc::new(ThreadPool::new(8)), + proc_count: Arc::new(AtomicU32::new(1)), + proc_queue: Arc::new(Mutex::new(vec![path])), + }) + } + + pub fn index(&self) -> Result<()> { + loop { + let pc = self.proc_count.load(Relaxed); + match pc { + 0 => break, + _ => { + let npath = { + let mut q = self.proc_queue.lock().unwrap(); + q.pop() + }; + + match npath { + None => continue, + Some(path) => { + self.proc_count.fetch_add(1, Relaxed); + let copy_of_queue = self.proc_queue.clone(); + let copy_of_files = self.files.clone(); + + self.threadpool.execute(move || { + if let Ok((files, dirs)) = Self::scan(path) { + let mut q = copy_of_queue.lock().unwrap(); + let mut f = copy_of_files.lock().unwrap(); + q.extend(files); + f.extend(dirs); + } + }); + } + } + } + } + } + + Ok(()) + } + + fn scan(scan_path: Box) -> Result<(Vec>, Vec>)> { + let mut files = vec![]; + let mut dirs = vec![]; + + fs::read_dir(scan_path.as_ref())? + .filter_map(Result::ok) + .for_each(|entry: fs::DirEntry| { + let entrymeta: fs::Metadata = entry.metadata().unwrap(); + let path = entry + .path() + .into_os_string() + .into_string() + .unwrap() + .into_boxed_str(); + + // TODO: Support Symlinks + if entrymeta.is_file() { + files.push(path); + } else if entrymeta.is_dir() { + dirs.push(path); + } + }); + + Ok((files, dirs)) + } +}