Channels added and threads no longer threading for days

This commit is contained in:
Jon Moroney 2018-02-09 15:18:00 +01:00
parent bf5d893ec9
commit 3844df72c7
2 changed files with 22 additions and 13 deletions

View file

@ -5,3 +5,5 @@ authors = ["Jon Moroney <j.moroney@cs.ru.nl>"]
[dependencies]
clap = "2.28.0"
threadpool = "*"
rayon = "0.9"

View file

@ -1,8 +1,10 @@
//Std imports
use std::io::Read;
use std::hash::Hash;
use std::io::BufReader;
use std::path::Path;
use std::thread;
use std::sync::mpsc::channel;
use std::collections::HashSet;
use std::path::PathBuf;
use std::cmp::Ordering;
@ -10,10 +12,15 @@ use std::fs::{self};
use std::collections::hash_map::DefaultHasher;
use std::hash::Hasher;
//External imports
extern crate clap;
extern crate threadpool;
extern crate rayon;
use clap::{Arg, App};
use rayon::prelude::*;
use threadpool::*;
#[derive(Debug, Clone)]
#[derive(Debug)]
struct Fileinfo{
file_hash: u64,
file_len: u64,
@ -86,20 +93,24 @@ fn main() {
let display_power = match arguments.value_of("Blocksize").unwrap_or(""){"K" => 1, "M" => 2, "G" => 3, _ => 0};
let blocksize = match arguments.value_of("Blocksize").unwrap_or(""){"K" => "Kilobytes", "M" => "Megabytes", "G" => "Gigabytes", _ => "Bytes"};
let display_divisor = 1024u64.pow(display_power);
let pool = ThreadPool::new(10);
let (sender, receiver) = channel();
let mut directory_results = Vec::new();
let mut thread_handles = Vec::new();
for arg in arguments.values_of("directories").unwrap().into_iter(){
let arg_str = String::from(arg);
thread_handles.push(thread::spawn(move|| -> Vec<Fileinfo> {
collect(Path::new(&arg_str), Vec::new())
let inner_sender = sender.clone();
thread_handles.push(pool.execute(move|| {
inner_sender.send(collect(Path::new(&arg_str), Vec::new())).unwrap();
}));
}
for handle in thread_handles {
directory_results.push(handle.join().unwrap());
}
pool.join();
directory_results.push(receiver.recv().unwrap());
// for handle in thread_handles {
// directory_results.push(pool.join());
// }
let mut complete_files: Vec<Fileinfo> = directory_results.into_iter().fold(Vec::new(), |mut unifier, element| {unifier.extend(element); unifier});
complete_files.sort_unstable();
//println!("complete_files len = {:?}", complete_files.len());
complete_files.dedup_by(|a, b| if a==b{
b.file_paths.extend(a.file_paths.drain());
true
@ -137,7 +148,7 @@ fn hash_file(file_path: &Path) -> Option<u64>{
fn collect(current_path: &Path, mut file_set: Vec<Fileinfo>) -> Vec<Fileinfo> {
if current_path.is_file(){
match hash_file(&current_path){
Some(hash_val) => {file_set.push(Fileinfo{file_paths: vec![current_path.to_path_buf()].into_iter().collect(), file_hash: hash_val, file_len: current_path.metadata().unwrap().len()})},
Some(hash_val) => {file_set.push(Fileinfo{file_paths: vec![current_path.to_path_buf()].into_par_iter().collect(), file_hash: hash_val, file_len: current_path.metadata().unwrap().len()})},
None => {println!("Error encountered hashing {:?}. Skipping.", current_path)}
};
return file_set
@ -148,16 +159,12 @@ fn collect(current_path: &Path, mut file_set: Vec<Fileinfo>) -> Vec<Fileinfo> {
Ok(paths) => for entry in paths {
match entry{
Ok(item) => {if item.file_type().ok().unwrap().is_dir(){
thread_handles.push(thread::spawn(move|| -> Vec<Fileinfo> {
collect(&item.path(), Vec::new())
}));
//file_set = collect(&item.path(), file_set);
} else if item.file_type().ok().unwrap().is_file(){
match hash_file(&item.path()){
Some(hash_val) => {file_set.push(Fileinfo{file_paths: vec![item.path()].into_iter().collect(), file_hash: hash_val, file_len: item.metadata().unwrap().len()})},
Some(hash_val) => {file_set.push(Fileinfo{file_paths: vec![item.path()].into_par_iter().collect(), file_hash: hash_val, file_len: item.metadata().unwrap().len()})},
None => {println!("Error encountered hashing {:?}. Skipping.", item.path())}
};
}