Add SharedQueuePool for part 4

Author: hindenbug

kvs-client-server/src/lib.rs

@@ -3,7 +3,7 @@ pub use engines::{KvStore, KvsEngine, SledKvsEngine};
 pub use error::{KvsError, Result};
 pub use network::Request;
 pub use server::Server;
-pub use thread_pool::{NaiveThreadPool, ThreadPool};
+pub use thread_pool::{NaiveThreadPool, SharedQueueThreadPool, ThreadPool};
 
 mod client;
 mod engines;

kvs-client-server/src/server.rs

@@ -1,5 +1,5 @@
 use crate::network::{GetResponse, RemoveResponse, Request, SetResponse};
-use crate::{KvsEngine, NaiveThreadPool, Result, ThreadPool};
+use crate::{KvsEngine, Result, SharedQueueThreadPool, ThreadPool};
 use log::{debug, error, info};
 use serde_json::Deserializer;
 use std::io::{BufReader, BufWriter, Write};
@@ -24,7 +24,7 @@ impl<E: KvsEngine> Server<E> {
 
     pub fn serve(&self) -> Result<()> {
         debug!("Waiting for connections...");
-        let thread_pool = NaiveThreadPool::new(1)?;
+        let thread_pool = SharedQueueThreadPool::new(num_cpus::get() as u32)?;
         let listnr = self.listener.try_clone().unwrap();
         for stream in listnr.incoming() {
             let engine = self.engine.clone();

kvs-client-server/src/thread_pool/mod.rs

@@ -1,8 +1,10 @@
 use crate::Result;
 
 mod naive;
+mod shared_queue;
 
 pub use self::naive::NaiveThreadPool;
+pub use self::shared_queue::SharedQueueThreadPool;
 
 pub trait ThreadPool {
     /// Creates a thread pool

kvs-client-server/src/thread_pool/naive.rs

@@ -1,7 +1,7 @@
-use super::ThreadPool;
-use crate::Result;
+use crate::{Result, ThreadPool};
 
 use std::thread;
+
 pub struct NaiveThreadPool;
 
 impl ThreadPool for NaiveThreadPool {

kvs-client-server/src/thread_pool/shared_queue.rs

@@ -0,0 +1,44 @@
+use crate::{Result, ThreadPool};
+use crossbeam::crossbeam_channel::{unbounded, Receiver, Sender};
+
+use std::thread;
+
+pub struct SharedQueueThreadPool(Sender<Box<dyn FnOnce() + Send + 'static>>);
+#[derive(Clone)]
+struct TaskReceiver(Receiver<Box<dyn FnOnce() + Send + 'static>>);
+
+impl ThreadPool for SharedQueueThreadPool {
+    fn new(threads: u32) -> Result<Self> {
+        let (sender, recv) = unbounded::<Box<dyn FnOnce() + Send + 'static>>();
+        for _ in 0..threads {
+            let r_clone = TaskReceiver(recv.clone());
+            thread::spawn(move || {
+                while let Ok(job) = r_clone.0.recv() {
+                    job();
+                }
+            });
+        }
+
+        Ok(SharedQueueThreadPool(sender))
+    }
+
+    fn spawn<F>(&self, job: F)
+    where
+        F: FnOnce() + Send + 'static,
+    {
+        self.0.send(Box::new(job)).unwrap();
+    }
+}
+
+impl Drop for TaskReceiver {
+    fn drop(&mut self) {
+        if thread::panicking() {
+            let r_clone = self.clone();
+            thread::spawn(move || {
+                while let Ok(job) = r_clone.0.recv() {
+                    job();
+                }
+            });
+        }
+    }
+}

kvs-client-server/tests/thread_pool.rs

@@ -2,7 +2,7 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
 use kvs::Result;
-use kvs::{NaiveThreadPool, ThreadPool};
+use kvs::{NaiveThreadPool, SharedQueueThreadPool, ThreadPool};
 
 use crossbeam_utils::sync::WaitGroup;
 
@@ -52,19 +52,19 @@ fn naive_thread_pool_spawn_counter() -> Result<()> {
     spawn_counter(pool)
 }
 
-// #[test]
-// fn shared_queue_thread_pool_spawn_counter() -> Result<()> {
-//     let pool = SharedQueueThreadPool::new(4)?;
-//     spawn_counter(pool)
-// }
+#[test]
+fn shared_queue_thread_pool_spawn_counter() -> Result<()> {
+    let pool = SharedQueueThreadPool::new(4)?;
+    spawn_counter(pool)
+}
 
 // #[test]
 // fn rayon_thread_pool_spawn_counter() -> Result<()> {
 //     let pool = RayonThreadPool::new(4)?;
 //     spawn_counter(pool)
 // }
 
-// #[test]
-// fn shared_queue_thread_pool_panic_task() -> Result<()> {
-//     spawn_panic_task::<SharedQueueThreadPool>()
-// }
+#[test]
+ fn shared_queue_thread_pool_panic_task() -> Result<()> {
+     spawn_panic_task::<SharedQueueThreadPool>()
+}