From d24a574405641b7770408401fc160be647f830af Mon Sep 17 00:00:00 2001
From: ridethepig <protonghost@163.com>
Date: Wed, 8 Mar 2023 11:27:13 +0800
Subject: [PATCH] Week6 done

---
 week6/parallel_map/Cargo.toml  |  2 +-
 week6/parallel_map/src/main.rs | 38 ++++++++++++++++++++++++++++++++++
 2 files changed, 39 insertions(+), 1 deletion(-)

diff --git a/week6/parallel_map/Cargo.toml b/week6/parallel_map/Cargo.toml
index 3d5d931..ef324b3 100644
--- a/week6/parallel_map/Cargo.toml
+++ b/week6/parallel_map/Cargo.toml
@@ -7,4 +7,4 @@ edition = "2018"
 # See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
 
 [dependencies]
-crossbeam-channel = "0.4.2"
\ No newline at end of file
+crossbeam-channel = "0.5.7"
\ No newline at end of file
diff --git a/week6/parallel_map/src/main.rs b/week6/parallel_map/src/main.rs
index b98cb3d..d789245 100644
--- a/week6/parallel_map/src/main.rs
+++ b/week6/parallel_map/src/main.rs
@@ -8,16 +8,54 @@ where
     U: Send + 'static + Default,
 {
     let mut output_vec: Vec<U> = Vec::with_capacity(input_vec.len());
+    for _ in 0..input_vec.len() {
+        output_vec.push(U::default());
+    }
     // TODO: implement parallel map!
+    let (distri_s, distri_r) = crossbeam_channel::unbounded();
+    let (collect_s , collect_r) = crossbeam_channel::unbounded();
+
+    let mut thread_handlers = Vec::new();
+    for _ in 0..num_threads {
+        let distri_r = distri_r.clone();
+        let collect_s = collect_s.clone();
+        let handler = thread::spawn(move || {
+            while let Ok((val, idx)) = distri_r.recv() {
+                let val = f(val);
+                collect_s.send((val, idx)).unwrap();
+            }
+        });
+        thread_handlers.push(handler);
+    }
+    drop(collect_s);
+    println!("spawned {} threads", num_threads);
+    let mut distri_counter = input_vec.len();
+    while let Some(val) = input_vec.pop() {
+        distri_counter -= 1;
+        distri_s.send((val, distri_counter)).unwrap();
+    }
+    drop(distri_s);
+    while let Ok((val, idx)) = collect_r.recv() {
+        output_vec[idx] = val;
+    }
+    println!("collected {} values", output_vec.len());
+    for handler in thread_handlers {
+        let _ = handler.join();
+    }
     output_vec
 }
 
 fn main() {
     let v = vec![6, 7, 8, 9, 10, 1, 2, 3, 4, 5, 12, 18, 11, 5, 20];
+    let mut squares_std = Vec::new();
+    for val in &v {
+        squares_std.push(val * val);
+    }
     let squares = parallel_map(v, 10, |num| {
         println!("{} squared is {}", num, num * num);
         thread::sleep(time::Duration::from_millis(500));
         num * num
     });
+    assert_eq!(squares_std, squares);
     println!("squares: {:?}", squares);
 }