Update solution for 2015 day 04

Implement simple thread pool and make sure that the smallest number is
always found.

2015/day-04/main.rs

@@ -1,4 +1,7 @@
 mod md5;
+mod threadpool;
+
+use threadpool::ThreadPool;
 
 macro_rules! infile {
     [$path:literal] => { ($path, include_str!($path)) }
@@ -9,86 +12,90 @@ const INPUTS: &[(&str, &str)] = &[
     infile!("inputs/puzzle.txt"),
 ];
 
-use std::sync::{Arc, atomic::{AtomicBool, Ordering}};
+fn extract_result(pool: &ThreadPool<Vec<usize>>) -> Option<usize> {
-
+    let mut smallest = None;
-fn solve_part_1(input: &str, range: (usize, usize), found: Arc<AtomicBool>) {
+    while let Some(result) = pool.next_result().unwrap() {
-    for i in range.0..range.1 {
+        let result = result.into_iter().min();
-        if found.load(Ordering::SeqCst) {
+        if !result.is_none() && (smallest.is_none() || smallest > result) {
-            break;
+            smallest = result;
         }
-        let secret = format!("{}{}", input, i);
+    }
-        let hexdigest = md5::hexdigest(&secret);
+    smallest
+}
 
-        if hexdigest.starts_with("00000") {
+fn solve_part_1(pool: &ThreadPool<Vec<usize>>, input: &str) -> usize {
-            found.store(true, Ordering::SeqCst);
+    let mut start = 1;
-            println!("5 zeroes found with hexdigest {} for {}: {}", hexdigest, secret, i);
+    let mut tries = 7;
-            break;
+
+    loop {
+        if tries == 0 {
+            match extract_result(&pool) {
+                None =>{ tries = 7 },
+                Some(v) => return v,
+            }
         }
+
+        let input = input.to_string();
+        pool.add_task(move || {
+            (start..).into_iter()
+                .take(20_000)
+                .filter_map(|i| {
+                    let secret = format!("{}{}", input, i);
+                    let hexdigest = md5::hexdigest(&secret);
+                    if hexdigest.starts_with("00000") { Some(i) } else { None }
+                })
+                .collect::<Vec<_>>()
+            });
+        tries -= 1;
+        start += 20_000;
     }
 }
 
-fn solve_part_2(input: &str, range: (usize, usize), found: Arc<AtomicBool>) {
+fn solve_part_2(pool: &ThreadPool<Vec<usize>>, input: &str) -> usize {
-    for i in range.0..range.1 {
+    let mut start = 1;
-        if found.load(Ordering::SeqCst) {
+    let mut tries = 8;
-            return;
-        }
-        let secret = format!("{}{}", input, i);
-        let hexdigest = md5::hexdigest(&secret);
 
-        if hexdigest.starts_with("000000") {
+    loop {
-            found.store(true, Ordering::SeqCst);
+        if tries == 0 {
-            println!("6 zeroes found with hexdigest {} for {}: {}", hexdigest, input, i);
+            match extract_result(&pool) {
-            break;
+                None =>{ tries = 7 },
+                Some(v) => return v,
+            }
         }
+
+        let input = input.to_string();
+        pool.add_task(move || {
+            (start..).into_iter()
+                .take(20_000)
+                .filter_map(|i| {
+                    let secret = format!("{}{}", input, i);
+                    let hexdigest = md5::hexdigest(&secret);
+                    if hexdigest.starts_with("000000") { Some(i) } else { None }
+                })
+                .collect::<Vec<_>>()
+            });
+
+        tries -= 1;
+        start += 20_000;
     }
 }
 
 fn main() {
-    use std::{thread, process};
+    let pool = ThreadPool::with_threads(8);
-
-    let mut threads = Vec::new();
 
     for (path, input) in INPUTS {
         println!("File: {}", path);
 
         for line in input.split('\n') {
-            if line.is_empty() {
+            if line.is_empty() { continue }
-                continue;
-            }
 
-            let found_solution_1 = Arc::new(AtomicBool::new(false));
+            println!("\tInput: {}", line);
-            let found_solution_2 = Arc::new(AtomicBool::new(false));
 
-            let ranges = (0..).into_iter().step_by(100_000).take(15).collect::<Vec<_>>();
+            let solution = solve_part_1(&pool, &line);
-            for window in ranges.windows(2) {
+            println!("\t\tFive leading zeroes: {}", solution);
-                let (from, to) = (window[0], window[1]);
-                println!("Searching solutions for {} in {}..{}", line, from, to);
 
-                let found = found_solution_1.clone();
+            let solution = solve_part_2(&pool, &line);
-                let t = thread::spawn(move || {
+            println!("\t\tSix leading zeroes: {}", solution);
-                    solve_part_1(&line, (from, to), found)
-                });
-                threads.push(t);
-            }
-
-            let ranges = (0..).into_iter().step_by(1_000_000).take(15).collect::<Vec<_>>();
-            for window in ranges.windows(2) {
-                let (from, to) = (window[0], window[1]);
-                println!("Searching solution for {} in {}..{}", line, from, to);
-
-                let found = found_solution_2.clone();
-                let t = thread::spawn(move || {
-                    solve_part_2(&line, (from, to), found)
-                });
-                threads.push(t);
-            }
-        }
-    }
-
-    for thread in threads {
-        if let Err(_) = thread.join() {
-            eprintln!("Failed to join thread");
-            process::exit(1);
         }
     }
 }

2015/day-04/threadpool.rs

@@ -0,0 +1,107 @@
+use std::{
+    thread,
+    sync::{Arc, Mutex, mpsc},
+};
+
+pub trait Task<O>
+    where O: Send + 'static
+{
+    fn process(&self) -> O;
+}
+
+impl<F, O> Task<O> for F
+    where F: Fn() -> O,
+          F: Send + 'static,
+          O: Send + 'static
+{
+    fn process(&self) -> O {
+        self()
+    }
+}
+
+type BoxTask<O> = Box<dyn Task<O> + Send + 'static>;
+type TaskReceiver<O> = Arc<Mutex<mpsc::Receiver<BoxTask<O>>>>;
+type TaskSender<O> = mpsc::Sender<BoxTask<O>>;
+type OutputSender<O> = Arc<Mutex<mpsc::Sender<O>>>;
+type OutputReceiver<O> = mpsc::Receiver<O>;
+
+#[derive(Clone)]
+struct Shared<O>
+    where O: Send + Clone + 'static
+{
+    incoming: TaskReceiver<O>,
+    output: OutputSender<O>,
+}
+
+pub struct ThreadPool<O>
+    where O: Send + Clone + 'static
+{
+    workers: Vec<thread::JoinHandle<()>>,
+    shared: Shared<O>,
+    tasks: TaskSender<O>,
+    results: OutputReceiver<O>,
+}
+
+impl<O> ThreadPool<O>
+    where O: Send + Clone + 'static
+{
+    pub fn with_threads(size: usize) -> Self {
+        let (tasks, incoming) = mpsc::channel();
+        let (output, results) = mpsc::channel();
+        let shared = Shared {
+            incoming: Arc::new(Mutex::new(incoming)),
+            output: Arc::new(Mutex::new(output)),
+        };
+        let mut pool = Self {
+            workers: Vec::new(),
+            shared,
+            tasks,
+            results
+        };
+
+        for worker_id in 0..size {
+            pool.spawn_worker(worker_id);
+        }
+
+        pool
+    }
+
+    pub fn add_task(&self, task: impl Task<O> + Send + 'static) {
+        self.tasks.send(Box::new(task)).expect("Failed to enqueue task");
+    }
+
+    pub fn next_result(&self) -> Result<Option<O>, mpsc::TryRecvError> {
+        match self.results.try_recv() {
+            Ok(output) => Ok(Some(output)),
+            Err(mpsc::TryRecvError::Empty) => Ok(None),
+            Err(e) => Err(e),
+        }
+    }
+
+    fn spawn_worker(&mut self, worker_id: usize) {
+        let shared = self.shared.clone();
+        let worker = thread::Builder::new()
+            .name(format!("worker-{}", worker_id))
+            .spawn(move || {
+                loop {
+                    let task = {
+                        let incoming = shared.incoming.lock().expect("Lock was poisoned");
+                        match incoming.recv() {
+                            Ok(task) => task,
+                            Err(_) => break,
+                        }
+                    };
+
+                    let output = task.process();
+
+                    let output_sender = shared.output.lock().expect("Lock was poisoned");
+                    if let Err(_) = output_sender.send(output) {
+                        break;
+                    }
+                }
+            })
+            .expect("Failed to spawn thread pool worker");
+
+        self.workers.push(worker);
+    }
+}