Concurrent queue working

hw10/task1/Cargo.toml

@@ -6,5 +6,9 @@ authors = ["Joshua Rutschmann <joshua.rutschmann@gmx.de>", "Lorenz Bung <lorenz.
 [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,12 +1,61 @@
 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::net::{TcpListener, TcpStream};
 use srv_commands::Command;
 use srv_config::Config;
-use HashServerError::Parse;
+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");
+        }
+    }
+}
 
 #[derive(Debug)]
 pub enum HashServerError {
@@ -26,7 +75,7 @@ impl From<srv_commands::ParseError> for HashServerError {
     }
 }
 
-fn handle_client(stream: &TcpStream, orders: &mut VecDeque<String>, v: bool) {
+fn handle_client(stream: &TcpStream, orders:OrderQueue<&str>) {
     let mut reader = BufReader::new(stream);
     let mut writer = BufWriter::new(stream);
 
@@ -38,32 +87,19 @@ fn handle_client(stream: &TcpStream, orders: &mut VecDeque<String>, v: bool) {
                     break;
                 }
 
-                let cmd = srv_commands::parse(&line).map_err(HashServerError::Parse);
+                let mut words = line.split_whitespace();
+                match words.next() {
+                    Some(difficulty) => {
 
-                match cmd {
-                    Ok(Command::Stage(str)) => {
-                        orders.push_front(str);
-                    }
-                    Ok(Command::Retrieve) => {
-                        if orders.is_empty() {
-                            let _ = writer.write(b"No order on stage!\n");
-
-                        } else {
-                            if let Some(latest_order) = orders.pop_front() {
-                                let _ = writer.write(latest_order.as_bytes());
-                                let _ = writer.write(b"\n");
-                            }
+                        if let Some(range) = words.next() {
                         }
-                    }
-                    Ok(Command::Control(ref control_string)) => {
-                        if v {
-                            println!("Received Control: {}", control_string);
+
+                        if let Some(port) = words.next() {
                         }
-                    }
-                    Err(Parse(e)) => {
-                        println!("Error occurred: {:?}", e);
-                    }
-                    _ => {}
+                        
+                        orders.add_order("Hello");
+                    },
+                    None => {},
                 }
 
                 let _ = writer.flush();
@@ -76,7 +112,6 @@ fn handle_client(stream: &TcpStream, orders: &mut VecDeque<String>, v: bool) {
 }
 
 pub fn main() {
-
     let c = Config::load();
 
     if c.verbosity > 0 {
@@ -92,30 +127,75 @@ pub fn main() {
 
     let host = format!("{}:{}", c.address, c.port);
 
-    let mut orders: VecDeque<String> = VecDeque::new();
-
-    if c.testing {
-        orders.push_front(String::from("Test3"));
-        orders.push_front(String::from("Test2"));
-        orders.push_front(String::from("Test1"));
-    }
-
-    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")
-                    }
-                    handle_client(&stream, &mut orders, c.verbosity > 0);
-                    if c.verbosity > 1 {
-                        println!("[DEBUG] Client diconnected")
+    let orders: OrderQueue<&str> = OrderQueue::new();
+    let mut results: VecDeque<Solution> = VecDeque::new();
+
+    let ord = 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();
+                        
+                        thread::spawn(move || {
+                            handle_client(&stream, order_queue);
+                        });
+
+                        if c.verbosity > 1 {
+                            println!("[DEBUG] Client diconnected")
+                        }
                     }
                 }
+            },
+            Err(e) => {
+                println!("Failed to start the MultiHashServer: {:?}", e);
             }
         }
-        Err(e) => {
-            println!("Failed to start the MultiHashServer: {:?}", e);
+    
+    });
+
+    let stdin = io::stdin();
+    let mut sl = stdin.lock();
+    let stdout = io::stdout();
+    let mut ol = stdout.lock();
+    
+    loop {
+        ol.write(b"HashServer> ");
+        ol.flush();
+        let mut line = String::new();
+        match sl.read_line(&mut line) {
+            Ok(_) => {
+                let mut lines = line.trim_right().split(' ');
+                match lines.next() {
+                    Some("exit") => process::exit(0),
+                    Some("halt") => {
+                        println!("Saving state...");
+                        //Serialize here
+                        process::exit(0);
+                    },
+                    Some("get") => {
+                        ord.serialize();
+                        println!("{:?}", ord);
+                    },
+                    Some("continue") => {
+                        println!("Restoring state")
+                    },
+                    Some("threads") => {
+                        if let Some(thr) = lines.next() {
+                            println!("Setting thread count to {}", thr);
+                        }
+                    },
+                    Some(&_) => {},
+                    None => {},
+                }
+            }
+            Err(_) => {},
         }
     }
 }

hw10/task1/srv-hasher/Cargo.toml

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

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

@@ -0,0 +1,309 @@
+extern crate sha2;
+extern crate time;
+
+#[cfg(feature = "SHA2")]
+use self::sha2::Sha256;
+use std::thread;
+use std::sync::Arc;
+use std::sync::atomic::AtomicBool;
+use std::sync::atomic::Ordering::Relaxed;
+use std::sync::mpsc::{Sender, channel};
+use time::{Duration, get_time};
+
+
+pub struct Sha256;
+
+pub trait Hasher {
+    type Output: HashResult;
+    fn hash(input: &[u8]) -> Self::Output;
+}
+
+pub trait HashResult {
+    /// Get the output in hex notation.
+    fn hex(&self) -> String;
+    /// Size of the output in bytes.
+    fn size() -> usize;
+}
+
+impl Hasher for Sha256 {
+    type Output = [u8; 32];
+
+    fn hash(input: &[u8]) -> Self::Output {
+        use self::sha2::*;
+        let mut tmp = Sha256::new();
+        tmp.input(input);
+        let r = tmp.result().as_slice().to_vec();
+        [
+            r[0],
+            r[1],
+            r[2],
+            r[3],
+            r[4],
+            r[5],
+            r[6],
+            r[7],
+            r[8],
+            r[9],
+            r[10],
+            r[11],
+            r[12],
+            r[13],
+            r[14],
+            r[15],
+            r[16],
+            r[17],
+            r[18],
+            r[19],
+            r[20],
+            r[21],
+            r[22],
+            r[23],
+            r[24],
+            r[25],
+            r[26],
+            r[27],
+            r[28],
+            r[29],
+            r[30],
+            r[31],
+        ]
+    }
+}
+
+impl HashResult for [u8; 32] {
+    fn hex(&self) -> String {
+        const HEX: [char; 16] = [
+            '0',
+            '1',
+            '2',
+            '3',
+            '4',
+            '5',
+            '6',
+            '7',
+            '8',
+            '9',
+            'a',
+            'b',
+            'c',
+            'd',
+            'e',
+            'f',
+        ];
+        let mut tmp = String::with_capacity(32 * 2);
+        for byte in self.iter() {
+            tmp.push(HEX[*byte as usize / 16]);
+            tmp.push(HEX[*byte as usize % 16]);
+        }
+        tmp
+    }
+
+    fn size() -> usize {
+        32
+    }
+}
+
+#[derive(Debug, PartialEq)]
+pub struct Solution {
+    pub number: usize,
+    pub hash: String,
+}
+
+/// Überprüft ob *base* und *number* auf einen Hash abbilden,
+/// der mit der übergebenen *difficulty* übereinstimmt.
+/// Falls ja, kommt eine `Solution` in der Option mit den Ergebnissen zurück.
+/// Falls nein, steht None im Optional
+pub fn verify_product(base: usize, number: usize, difficulty: &String) -> Option<Solution> {
+    let sol = base * number;
+    let input = sol.to_string();
+    let bytes = input.as_bytes();
+
+    let hash = Sha256::hash(bytes).hex();
+
+    if hash.ends_with(difficulty) {
+        return Some(Solution {
+            number: number,
+            hash: hash,
+        });
+    }
+
+    None
+}
+
+/// 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_hash(
+    hash: &String,
+    base: usize,
+    start: 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 {
+
+        // Der special Parameter begrenzt die Anzahl der load()-Aufrufe auf jeden n.ten Loop.
+        if n % special == 0 && found.load(Relaxed) {
+            // Anderer Thread hat eine Lösung gefunden (sync ist aktiviert). Beende Suche.
+            break;
+        }
+
+        if let Some(solution) = verify_product(base, n, hash) {
+            if sync {
+                found.store(true, Relaxed);
+            }
+            // Sende gefundene Solution an den Consumer.
+            let _ = solution_tx.send(solution);
+            // Beende Suche.
+            break;
+        }
+        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.
+/// Gibt ggf. die Solution (für die Tests) zurück.
+pub fn search_with_threads(
+    threads: usize,
+    diff_string: String,
+    with_base: usize,
+    time_measurement: bool,
+    verbosity: u64,
+    sync: Option<usize>,
+    wait: bool,
+) -> 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 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 || {
+
+                // Suche in jedem der Threads.
+                search_hash(
+                    &diff,
+                    with_base,
+                    i,
+                    threads,
+                    sync.is_some(),
+                    found,
+                    sync.unwrap_or(1),
+                    solution_tx,
+                    timing_tx,
+                    m,
+                );
+
+                if verbosity > 1 {
+                    println!("[DEBUG] Thread {} exited", i);
+                }
+            }));
+        }
+    } else {
+        // Suche auf dem Main-Thread.
+        search_hash(
+            &diff,
+            with_base,
+            0,
+            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.
+    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
+}
+