Everything working except from deserialization.

hw10/task1/Cargo.toml

@@ -4,11 +4,7 @@ version = "0.2.0"
 authors = ["Joshua Rutschmann <joshua.rutschmann@gmx.de>", "Lorenz Bung <lorenz.bung@htwg-konstanz.de>"]
 
 [dependencies]
-srv-commands = { path = "srv-commands" }
 srv-config = { path = "srv-config" }
 srv-hasher = { path = "srv-hasher" }
-serde = "1.0.27"
-serde_derive = "1"
-serde_json = "1"
 
 [workspace]

hw10/task1/src/main.rs

@@ -1,83 +1,22 @@
-extern crate srv_commands;
 extern crate srv_config;
 extern crate srv_hasher;
 
-#[macro_use]
-extern crate serde_derive;
-extern crate serde;
-extern crate serde_json;
-
-use std::io::{BufReader, BufRead, BufWriter, Write};
-use std::collections::VecDeque;
+use std::io::{BufReader, BufRead, Write};
 use std::net::{TcpListener, TcpStream};
-use srv_commands::Command;
+use std::sync::atomic::{AtomicUsize, Ordering};
 use srv_config::Config;
 use srv_hasher::Solution;
 use std::{process, io, thread};
 use std::sync::Arc;
-use std::sync::Mutex;
-
-//#[derive(Serialize, Deserialize)]
-#[derive(Clone, Debug)]
-struct OrderQueue<T: Send + Copy> {
-    inner: Arc<Mutex<VecDeque<T>>>,
-}
-
-impl<T: Send + Copy> OrderQueue<T> {
-
-    fn new() -> Self {
-        Self { inner: Arc::new(Mutex::new(VecDeque::new())) }
-    }
-
-    fn deserialize() {
-        
-    }
-
-    fn serialize(&self) {
-        let inner_queue = self.inner.lock();
-        if let Ok(mut queue) = inner_queue {
-        }
-    }
-    
-    fn get_latest_order(&self) -> Option<T> {
-        let inner_queue = self.inner.lock();
-        if let Ok(mut queue) = inner_queue {
-            queue.pop_front()
-        } else {
-            panic!("MutexError");
-        }
-    }
-
-    fn add_order(&self, order: T) {
-        if let Ok(mut queue) = self.inner.lock() {
-            queue.push_front(order);
-        } else {
-            panic!("MutexError");
-        }
-    }
-}
+use srv_hasher::ParallelQueue;
 
 #[derive(Debug)]
 pub enum HashServerError {
-    Parse(srv_commands::ParseError),
     Io(std::io::Error),
 }
 
-impl From<std::io::Error> for HashServerError {
-    fn from(err: std::io::Error) -> HashServerError {
-        HashServerError::Io(err)
-    }
-}
-
-impl From<srv_commands::ParseError> for HashServerError {
-    fn from(err: srv_commands::ParseError) -> HashServerError {
-        HashServerError::Parse(err)
-    }
-}
-
-fn handle_client(stream: &TcpStream, orders:OrderQueue<&str>) {
+fn handle_client(stream: &TcpStream, orders:ParallelQueue<String>) {
     let mut reader = BufReader::new(stream);
-    let mut writer = BufWriter::new(stream);
 
     loop {
         let mut line = String::new();
@@ -86,23 +25,7 @@ fn handle_client(stream: &TcpStream, orders:OrderQueue<&str>) {
                 if n == 0 {
                     break;
                 }
-
-                let mut words = line.split_whitespace();
-                match words.next() {
-                    Some(difficulty) => {
-
-                        if let Some(range) = words.next() {
-                        }
-
-                        if let Some(port) = words.next() {
-                        }
-                        
-                        orders.add_order("Hello");
-                    },
-                    None => {},
-                }
-
-                let _ = writer.flush();
+                orders.add(line.trim_right().to_string());
             }
             _ => {
                 break;
@@ -127,29 +50,19 @@ pub fn main() {
 
     let host = format!("{}:{}", c.address, c.port);
 
-    let orders: OrderQueue<&str> = OrderQueue::new();
-    let mut results: VecDeque<Solution> = VecDeque::new();
-
-    let ord = orders.clone();
-
+    let orders: ParallelQueue<String> = ParallelQueue::new();
+	
+	let orders_1 = orders.clone();
+	
     thread::spawn(move || {
         match TcpListener::bind(host).map_err(HashServerError::Io) {
             Ok(listener) => {
                 for s in listener.incoming() {
                     if let Ok(stream) = s {
-                        if c.verbosity > 1 {
-                            println!("[DEBUG] New Client connected");
-                        }                       
-                        
-                        let order_queue = orders.clone();
-                        
+                        let order_queue = orders_1.clone();
                         thread::spawn(move || {
                             handle_client(&stream, order_queue);
                         });
-
-                        if c.verbosity > 1 {
-                            println!("[DEBUG] Client diconnected")
-                        }
                     }
                 }
             },
@@ -159,15 +72,81 @@ pub fn main() {
         }
     
     });
-
-    let stdin = io::stdin();
-    let mut sl = stdin.lock();
-    let stdout = io::stdout();
-    let mut ol = stdout.lock();
     
+	let threads_arc = Arc::new(AtomicUsize::new(5));
+	
+	let orders_2 = orders.clone();
+	
+	let threads = threads_arc.clone();
+	thread::spawn(move || {
+		loop {
+			if let Some(order) = orders_2.pop() {
+				let mut words = order.split(',');
+				let t = threads.load(Ordering::SeqCst);
+				
+				match (words.next(),words.next(),words.next()){
+					(Some(difficulty), None, None) => {
+						println!("\nSearching with {} threads for hash with difficulty: {}", t, difficulty);
+						prompt();
+						if let Some(sol) = srv_hasher::search_with_threads(t, difficulty.to_string(), 42, <usize>::max_value(), Some(1), true) {
+							println!("\nFound solution {:?}", sol);
+							prompt();
+						}
+					},
+					(Some(difficulty), Some(range), None) => {
+						if let Ok(rng) = range.parse::<usize>() {
+							println!("\nSearching with {} threads for hash with difficulty ({}) in range 0 - {} ", t, difficulty, range);
+							prompt();
+							let result_queue = srv_hasher::search_multiple_with_threads(t, difficulty.to_string(), 42, rng);
+							println!();
+							while let Some(solution) = result_queue.pop() {
+								println!("{:?}", solution);
+							}
+						}
+					},
+					(Some(difficulty), Some(range), Some(port)) => {
+						if let Ok(rng) = range.parse::<usize>() {
+							println!("\nSearching with {} threads for hash with difficulty ({}) in range 0 - {} ", t, difficulty, range);
+							prompt();
+
+							let result_queue = srv_hasher::search_multiple_with_threads(t, difficulty.to_string(), 42, rng);
+							let host = format!("{}:{}", c.address, port);
+							thread::spawn(move || {
+								match TcpListener::bind(host) {
+									Ok(listener) => {
+										for s in listener.incoming() {
+											if let Ok(mut stream) = s {
+												while let Some(solution) = result_queue.pop() {
+													stream.write(format!("{:?}\n", solution).as_bytes());
+													stream.flush();
+												}
+												
+											}
+										}
+									},
+									Err(e) => {
+										println!("Failed to start the MultiHashServer: {:?}", e);
+									}
+								}
+							});
+						}
+					},
+					(_, _, _) => {
+					
+					}
+				}
+			}
+		}
+	});
+	
+	
+	let orders_3 = orders.clone();
+	
+	let stdin = io::stdin();
+    let mut sl = stdin.lock();
+	
     loop {
-        ol.write(b"HashServer> ");
-        ol.flush();
+		prompt();
         let mut line = String::new();
         match sl.read_line(&mut line) {
             Ok(_) => {
@@ -177,18 +156,19 @@ pub fn main() {
                     Some("halt") => {
                         println!("Saving state...");
                         //Serialize here
+                        orders_3.serialize();
                         process::exit(0);
                     },
-                    Some("get") => {
-                        ord.serialize();
-                        println!("{:?}", ord);
-                    },
                     Some("continue") => {
-                        println!("Restoring state")
+                        println!("Restoring state");
+                        //orders.deserialize();
                     },
                     Some("threads") => {
-                        if let Some(thr) = lines.next() {
-                            println!("Setting thread count to {}", thr);
+                        if let Some(thread_string) = lines.next() {
+							if let Ok(thr) = thread_string.parse::<usize>() {
+								println!("Setting thread count to {}", thr);
+								threads_arc.store(thr, Ordering::SeqCst);
+							}
                         }
                     },
                     Some(&_) => {},
@@ -199,3 +179,8 @@ pub fn main() {
         }
     }
 }
+
+fn prompt(){
+	print!("HashServer> ");
+	io::stdout().flush().ok().expect("Could not flush stdout");
+}

hw10/task1/srv-commands/Cargo.toml

@@ -1,8 +0,0 @@
-[package]
-name = "srv-commands"
-version = "0.1.0"
-authors = ["Joshua Rutschmann <joshua.rutschmann@gmx.de>"]
-
-[dependencies]
-sha2 = "0.7.0"
-time = "0.1"

hw10/task1/srv-commands/src/lib.rs

@@ -1,39 +0,0 @@
-pub fn parse(message: &str) -> Result<Command, ParseError> {
-    let m: String = String::from(message.trim_right());
-    let mut line = m.lines();
-    match line.next() {
-        Some(x) => {
-            let mut str = x.split_whitespace();
-            match str.next() {
-                Some("STAGE") => {
-                    let msg = m[6..].trim_left();
-                    Ok(Command::Stage(msg.to_string()))
-                }
-                Some("CONTROL") => {
-                    let cmd = m[8..].trim_left();
-                    Ok(Command::Control(cmd.to_string()))
-                }
-                Some("RETRIEVE") => Ok(Command::Retrieve),
-                Some(_) => Err(ParseError(ErrorType::UnknownCommand)),
-                None => Err(ParseError(ErrorType::EmptyString)),
-            }
-        }
-        None => Err(ParseError(ErrorType::EmptyString)),
-    }
-}
-
-#[derive(Debug, PartialEq)]
-pub enum Command {
-    Stage(String),
-    Control(String),
-    Retrieve,
-}
-
-#[derive(Debug, PartialEq)]
-pub struct ParseError(pub ErrorType);
-
-#[derive(Debug, PartialEq)]
-pub enum ErrorType {
-    UnknownCommand,
-    EmptyString,
-}

hw10/task1/srv-commands/tests/srv-commands.rs

@@ -1,43 +0,0 @@
-#[cfg(test)]
-
-mod tests {
-	extern crate srv_commands;
-    use self::srv_commands::*;
-
-    #[test]
-    fn empty_returns_correct_command() {
-        assert_eq!(parse("\n"), Err(ParseError(ErrorType::EmptyString)))
-    }
-
-    #[test]
-    fn not_known_command_returns_correct_command() {
-        assert_eq!(parse("Hello\n"), Err(ParseError(ErrorType::UnknownCommand)))
-    }
-
-    #[test]
-    fn stage_returns_correct_command() {
-        assert_eq!(
-            parse("STAGE Hello\n"),
-            Ok(Command::Stage("Hello".to_string()))
-        )
-    }
-
-    #[test]
-    fn control_returns_correct_command() {
-        assert_eq!(
-            parse("CONTROL Hello\n"),
-            Ok(Command::Control("Hello".to_string()))
-        )
-    }
-
-    #[test]
-    fn retrieve_returns_correct_command() {
-        assert_eq!(parse("RETRIEVE\n"), Ok(Command::Retrieve))
-    }
-
-    #[test]
-    fn retrieve_with_arguments_returns_correct_command() {
-        assert_eq!(parse("RETRIEVE Hello\n"), Ok(Command::Retrieve))
-    }
-
-}

hw10/task1/srv-hasher/Cargo.toml

@@ -5,4 +5,6 @@ authors = ["Joshua Rutschmann <joshua.rutschmann@gmx.de>"]
 
 [dependencies]
 sha2 = "0.7.0"
-time = "0.1"
+serde = "1.0.27"
+serde_derive = "1"
+serde_json = "1"

hw10/task1/srv-hasher/src/lib.rs

@@ -1,14 +1,73 @@
 extern crate sha2;
-extern crate time;
+
+#[macro_use]
+extern crate serde_derive;
+extern crate serde;
+extern crate serde_json;
 
 #[cfg(feature = "SHA2")]
 use self::sha2::Sha256;
 use std::thread;
-use std::sync::Arc;
+use std::sync::{Arc, Mutex};
+use std::collections::VecDeque;
 use std::sync::atomic::AtomicBool;
 use std::sync::atomic::Ordering::Relaxed;
 use std::sync::mpsc::{Sender, channel};
-use time::{Duration, get_time};
+use std::fs::File;
+use std::io::Write;
+use std::ops::Deref;
+
+#[derive(Serialize)]
+struct InnerQueue<T> {
+	inner: VecDeque<T>,
+}
+
+impl<'a, T: Send + Clone + serde::Serialize> InnerQueue<T> {
+
+    fn new() -> Self {
+        Self { inner: VecDeque::new() }
+    }
+	
+}
+
+#[derive(Clone)]
+pub struct ParallelQueue<T> {
+    inner: Arc<Mutex<InnerQueue<T>>>,
+}
+
+impl<'a, T: Send + Clone + serde::Serialize> ParallelQueue<T> {
+
+    pub fn new() -> Self {
+        Self { inner: Arc::new(Mutex::new(InnerQueue::new())) }
+    }
+
+    pub fn serialize(&self) {
+        let inner_queue = self.inner.lock();
+        if let Ok(queue) = inner_queue {
+            let queue_json = serde_json::to_string(queue.deref()).unwrap();
+            let mut file = File::create("orders.save").unwrap();
+            file.write_all(queue_json.as_bytes());
+        }
+    }
+    
+    pub fn pop(&self) -> Option<T> {
+        let inner_queue = self.inner.lock();
+        if let Ok(mut queue) = inner_queue {
+            queue.inner.pop_front()
+        } else {
+            panic!("MutexError");
+        }
+    }
+	
+    pub fn add(&self, order: T) {
+        if let Ok(mut queue) = self.inner.lock() {
+            queue.inner.push_back(order);
+        } else {
+            panic!("MutexError");
+        }
+    }
+}
+
 
 
 pub struct Sha256;
@@ -103,7 +162,7 @@ impl HashResult for [u8; 32] {
     }
 }
 
-#[derive(Debug, PartialEq)]
+#[derive(Debug, PartialEq, Clone, Serialize)]
 pub struct Solution {
     pub number: usize,
     pub hash: String,
@@ -137,19 +196,16 @@ fn search_hash(
     hash: &String,
     base: usize,
     start: usize,
+	end: usize,
     total: usize,
     sync: bool,
     found: Arc<AtomicBool>,
     special: usize,
     solution_tx: Sender<Solution>,
-    timing_tx: Sender<(Duration, usize)>,
-    measure: bool,
 ) {
-    let max = <usize>::max_value();
     let mut n = start;
 
-    let thread_start = get_time();
-    while n < max {
+    while n < end {
 
         // Der special Parameter begrenzt die Anzahl der load()-Aufrufe auf jeden n.ten Loop.
         if n % special == 0 && found.load(Relaxed) {
@@ -168,12 +224,6 @@ fn search_hash(
         }
         n += total;
     }
-
-    // Falls measure aktiviert ist Sende Zeitdauer und Anzahl Loops an den Consumer.
-    if measure {
-        let thread_end = get_time();
-        let _ = timing_tx.send((thread_end - thread_start, (n / total)));
-    }
 }
 
 /// Teilt, wenn nötig, die Suche nach dem Hash auf mehrere Threads auf.
@@ -182,35 +232,25 @@ pub fn search_with_threads(
     threads: usize,
     diff_string: String,
     with_base: usize,
-    time_measurement: bool,
-    verbosity: u64,
+	range_of_numbers: usize,
     sync: Option<usize>,
     wait: bool,
-) -> Option<Solution> {
-
+) -> Option<Solution> {	
     let diff = Arc::new(diff_string);
     let mut children = vec![];
     let mut solution = None;
 
-    let (solution_tx, solution_rx) = channel();
-    let (timing_tx, timing_rx) = channel();
-
+	let (solution_tx, solution_rx) = channel();
+	
     let found = Arc::new(AtomicBool::new(false));
-    let m = time_measurement && verbosity > 0;
 
-    let total_start = get_time();
     if threads > 1 {
 
-        if verbosity > 0 {
-            println!("Searching with {} threads", threads);
-        }
-
         // Erstellt Anzahl angeforderter Threads.
         // Klont für jeden Thread die Referenz auf die gemeinsamen Variablen.
         for i in 0..threads {
             let diff = diff.clone();
             let solution_tx = solution_tx.clone();
-            let timing_tx = timing_tx.clone();
             let found = found.clone();
 
             children.push(thread::spawn(move || {
@@ -220,18 +260,14 @@ pub fn search_with_threads(
                     &diff,
                     with_base,
                     i,
+					range_of_numbers,
                     threads,
                     sync.is_some(),
                     found,
                     sync.unwrap_or(1),
                     solution_tx,
-                    timing_tx,
-                    m,
                 );
 
-                if verbosity > 1 {
-                    println!("[DEBUG] Thread {} exited", i);
-                }
             }));
         }
     } else {
@@ -240,70 +276,116 @@ pub fn search_with_threads(
             &diff,
             with_base,
             0,
+			range_of_numbers,
             1,
             sync.is_some(),
             found,
             sync.unwrap_or(1),
             solution_tx,
-            timing_tx,
-            m,
         );
 
-        if verbosity > 1 {
-            println!("[DEBUG] Finished search on main thread.");
-        }
     }
-
-    // Empfängt die Lösung von einem der Producer.
+	
+	// Empfängt die Lösung von einem der Producer.
     match solution_rx.recv() {
         Ok(sol) => {
             solution = Some(Solution {
                 number: sol.number,
                 hash: sol.hash.clone(),
             });
-            let total_end = get_time();
-            println!("Number: {} --> hash: {}", sol.number, sol.hash);
-            if time_measurement && verbosity == 0 {
-                let diff = total_end - total_start;
-                let s = diff.num_seconds();
-                let ms = diff.num_milliseconds();
-                let us = diff.num_microseconds().unwrap_or(ms * 1000);
-                println!("(Duration {}s / {}ms / {}us)", s, ms, us);
-            }
         }
         Err(_) => {}
     }
 
-    let mut sum_loops = 0usize;
-    let mut sum_time: Duration = Duration::zero();
-
     for child in children {
-
-        if time_measurement && verbosity > 0 {
-            // Warte auf Zeitstatistik-Nachricht von jedem Thread auf dem zweiten MPSC Channel.
-            if let Ok(stats) = timing_rx.recv() {
-                // Addiere Werte dieses threads zu der Summe.
-                sum_time = sum_time + stats.0;
-                sum_loops += stats.1;
-            }
-        }
-
         // Falls *wait* wahr ist, warte auf aktuellen thread (child)
         if wait {
             let _ = child.join();
         }
     }
 
-    // Gebe die Gesamtergebnisse der Threads aus.
-    if time_measurement && verbosity > 0 {
-        println!("Sum Loops in Producers:       {}", sum_loops);
-        let s = sum_time.num_seconds();
-        let ms = sum_time.num_milliseconds();
-        let us = sum_time.num_microseconds().unwrap_or(ms * 1000);
-        println!("Sum Duration in Producers:    {}s / {}ms / {}us", s, ms, us);
-    }
-
     // Gebe die Option einer Solution zurück.
     solution
 }
 
+
+
+/// Sucht nach einem Hash für die angegebene Basis und die Schwierigkeit.
+/// Wenn `total` > 1 ist, dann hat jeder Aufruf mit einem anderen `start`-Wert (von 0 - total)
+/// eine disjunkte Zahlenmenge für die Suche zur Auswahl.
+fn search_multiple_hash(
+    hash: &String,
+    base: usize,
+    start: usize,
+	end: usize,
+    total: usize,
+	queue: ParallelQueue<Solution>,
+) {
+    let mut n = start;
+
+    while n < end {
+        if let Some(solution) = verify_product(base, n, hash) {
+
+            // Sende gefundene Solution an den Consumer.
+			queue.add(solution);
+            break;
+        }
+        n += total;
+    }
+}
+
+/// Teilt, wenn nötig, die Suche nach dem Hash auf mehrere Threads auf.
+/// Gibt ggf. die Solution (für die Tests) zurück.
+pub fn search_multiple_with_threads(
+    threads: usize,
+    diff_string: String,
+    with_base: usize,
+	range_of_numbers: usize,
+) -> ParallelQueue<Solution> {	
+    let diff = Arc::new(diff_string);
+    let mut children = vec![];
+	let result_queue:ParallelQueue<Solution> = ParallelQueue::new();
+
+    if threads > 1 {
+
+        // Erstellt Anzahl angeforderter Threads.
+        // Klont für jeden Thread die Referenz auf die gemeinsamen Variablen.
+        for i in 0..threads {
+            let diff = diff.clone();
+			let queue = result_queue.clone();
+			
+            children.push(thread::spawn(move || {
+
+                // Suche in jedem der Threads.
+                search_multiple_hash(
+                    &diff,
+                    with_base,
+                    i,
+					range_of_numbers,
+                    threads,
+                    queue,
+                );
+
+            }));
+        }
+    } else {
+        // Suche auf dem Main-Thread.
+		let queue = result_queue.clone();
+        search_multiple_hash(
+            &diff,
+            with_base,
+            0,
+			range_of_numbers,
+            1,
+			queue,
+		);
+
+    }
+
+    for child in children {
+        let _ = child.join();
+    }
+	
+	result_queue
+}
+