From 3844df72c704982e4fb1bcb2328bbda0e9417b97 Mon Sep 17 00:00:00 2001
From: Jon Moroney <j.moroney@cs.ru.nl>
Date: Fri, 9 Feb 2018 15:18:00 +0100
Subject: [PATCH] Channels added and threads no longer threading for days

---
 Cargo.toml  |  2 ++
 src/main.rs | 33 ++++++++++++++++++++-------------
 2 files changed, 22 insertions(+), 13 deletions(-)

diff --git a/Cargo.toml b/Cargo.toml
index 5bcb72b..1c00480 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -5,3 +5,5 @@ authors = ["Jon Moroney <j.moroney@cs.ru.nl>"]
 
 [dependencies]
 clap = "2.28.0"
+threadpool = "*"
+rayon = "0.9"
diff --git a/src/main.rs b/src/main.rs
index d7bc879..7d4c122 100644
--- a/src/main.rs
+++ b/src/main.rs
@@ -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())}
                         };
                     }